收集 Keeper Enterprise 安全性記錄

支援的國家/地區:

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

Keeper Enterprise Security 是企業密碼管理工具和權限存取管理平台,可保護憑證、密鑰和機密資料。透過 Reporting and Alerts API,這項服務會產生保險箱存取、密碼變更、共用事件和管理動作的稽核記錄。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 GCP 專案
  • 建立及管理 GCS bucket 的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • 具備管理員權限的 Keeper 管理控制台特殊存取權
  • 啟用進階報表和快訊模組 (ARAM) 的 Keeper Enterprise 或 Business 方案

建立 Google Cloud Storage bucket

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

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

收集 Keeper Enterprise Security API 憑證

建立 API 金鑰

  1. 登入 Keeper 管理控制台
  2. 依序前往「管理」>「報表和快訊」
  3. 按一下「API 金鑰」,或前往 API 金鑰管理部分。
  4. 點選「建立 API 金鑰」
  5. 輸入 API 金鑰名稱 (例如 Google Security Operations Integration)。
  6. 複製下列詳細資料並儲存到安全位置:
    • API 金鑰:產生的 API 金鑰值
    • 私密金鑰:下載私密金鑰檔案 (用於產生權杖)
  7. 從管理控制台記下 Keeper Enterprise Enterprise ID

判斷 API 基準網址

  • Keeper API 基本網址取決於資料中心區域:

    區域 API 基礎網址
    美國 https://keepersecurity.com
    歐盟 https://keepersecurity.eu
    AU https://keepersecurity.com.au
    CA https://keepersecurity.ca
    JP https://keepersecurity.jp
    美國 GovCloud https://govcloud.keepersecurity.us

驗證權限

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

  1. 登入 Keeper 管理控制台
  2. 依序前往「管理」>「報表和快訊」
  3. 如果能看到「報表和快訊」資訊主頁和「API 金鑰」部分,表示您具備必要權限。
  4. 如果沒有看到這個選項,請與 Keeper 管理員聯絡,啟用 ARAM 模組並授予管理員存取權。

測試 API 存取權

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

    # Replace with your actual credentials
    KEEPER_API_KEY="your-api-key"
    KEEPER_BASE="https://keepersecurity.com"
    
    # Test API access - get audit events (requires signed token)
    # Note: Keeper uses a signed JWT for authentication.
    # Refer to Keeper Commander CLI for testing:
    keeper audit-report --format json --limit 1
    

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

建立 Cloud Run 函式來收集記錄

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

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

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

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

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

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

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

    1. 按一下「變數與密鑰」
    2. 針對每個環境變數,按一下「+ 新增變數」
    變數名稱 範例值 說明
    GCS_BUCKET keeper-audit-logs GCS bucket 名稱
    GCS_PREFIX keeper 記錄檔的前置字串
    STATE_KEY keeper/state.json 狀態檔案路徑
    KEEPER_API_BASE https://keepersecurity.com Keeper API 基本網址
    KEEPER_API_KEY your-api-key Keeper API 金鑰
    KEEPER_PRIVATE_KEY base64-encoded-private-key Base64 編碼的私密金鑰
    KEEPER_ENTERPRISE_ID 12345 企業 ID
    MAX_RECORDS 5000 每次執行的記錄數上限
    PAGE_SIZE 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 json
        import os
        import urllib3
        from datetime import datetime, timezone, timedelta
        import time
        import base64
        import hashlib
        import hmac
      
        # 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', 'keeper')
        STATE_KEY = os.environ.get('STATE_KEY', 'keeper/state.json')
        KEEPER_API_BASE = os.environ.get('KEEPER_API_BASE', 'https://keepersecurity.com')
        KEEPER_API_KEY = os.environ.get('KEEPER_API_KEY')
        KEEPER_PRIVATE_KEY = os.environ.get('KEEPER_PRIVATE_KEY')
        KEEPER_ENTERPRISE_ID = os.environ.get('KEEPER_ENTERPRISE_ID')
        MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
        PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
        LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
        def to_unix_millis(dt: datetime) -> int:
          """Convert datetime to Unix epoch milliseconds."""
          if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
          dt = dt.astimezone(timezone.utc)
          return int(dt.timestamp() * 1000)
      
        def parse_datetime(value: str) -> datetime:
          """Parse ISO datetime string to datetime object."""
          if value.endswith("Z"):
            value = value[:-1] + "+00:00"
          return datetime.fromisoformat(value)
      
        def get_auth_token():
          """
          Generate authentication token for Keeper Reporting API.
          Uses HMAC-SHA512 signing with the private key.
          """
          api_base = KEEPER_API_BASE.rstrip('/')
          token_url = f"{api_base}/api/rest/enterprise/auth/token"
      
          # Build token request
          timestamp = str(int(time.time()))
          message = f"{KEEPER_API_KEY}:{timestamp}"
      
          # Decode private key from base64
          private_key_bytes = base64.b64decode(KEEPER_PRIVATE_KEY)
      
          # Sign with HMAC-SHA512
          signature = hmac.new(
            private_key_bytes,
            message.encode('utf-8'),
            hashlib.sha512
          ).digest()
          signature_b64 = base64.b64encode(signature).decode('utf-8')
      
          body = json.dumps({
            'api_key': KEEPER_API_KEY,
            'timestamp': timestamp,
            'signature': signature_b64,
            'enterprise_id': KEEPER_ENTERPRISE_ID
          })
      
          headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
          }
      
          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 auth token: {response.status} - {response.data.decode('utf-8')}")
      
            data = json.loads(response.data.decode('utf-8'))
            return data.get('token', data.get('access_token'))
      
          raise RuntimeError("Failed to get auth token after 3 retries")
      
        @functions_framework.cloud_event
        def main(cloud_event):
          """
          Cloud Run function triggered by Pub/Sub to fetch Keeper Enterprise
          Security audit logs and write to GCS.
      
          Args:
            cloud_event: CloudEvent object containing Pub/Sub message
          """
      
          if not all([GCS_BUCKET, KEEPER_API_KEY, KEEPER_PRIVATE_KEY, KEEPER_ENTERPRISE_ID]):
            print('Error: Missing required environment variables')
            return
      
          try:
            bucket = storage_client.bucket(GCS_BUCKET)
      
            # Load state
            state = load_state(bucket, STATE_KEY)
      
            # Determine time window
            now = datetime.now(timezone.utc)
            last_time = None
      
            if isinstance(state, dict) and state.get("last_event_time"):
              try:
                last_time = parse_datetime(state["last_event_time"])
                # Overlap by 2 minutes to catch any delayed events
                last_time = last_time - timedelta(minutes=2)
              except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")
      
            if last_time is None:
              last_time = now - timedelta(hours=LOOKBACK_HOURS)
      
            print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
      
            # Get auth token
            token = get_auth_token()
      
            # Fetch logs
            records, newest_event_time = fetch_logs(
              token=token,
              start_time=last_time,
              end_time=now,
              page_size=PAGE_SIZE,
              max_records=MAX_RECORDS,
            )
      
            if not records:
              print("No new log records found.")
              save_state(bucket, STATE_KEY, now.isoformat())
              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 records]) + '\n'
            blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
            print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
      
            # Update state with newest event time
            if newest_event_time:
              save_state(bucket, STATE_KEY, newest_event_time)
            else:
              save_state(bucket, STATE_KEY, now.isoformat())
      
            print(f"Successfully processed {len(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():
              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, last_event_time_iso: str):
          """Save the last event timestamp to GCS state file."""
          try:
            state = {'last_event_time': last_event_time_iso}
            blob = bucket.blob(key)
            blob.upload_from_string(
              json.dumps(state, indent=2),
              content_type='application/json'
            )
            print(f"Saved state: last_event_time={last_event_time_iso}")
          except Exception as e:
            print(f"Warning: Could not save state: {e}")
      
        def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
          """
          Fetch audit event logs from Keeper Reporting and Alerts API
          with pagination and rate limiting.
      
          Args:
            token: Authentication token
            start_time: Start time for log query
            end_time: End time for log query
            page_size: Number of records per page
            max_records: Maximum total records to fetch
      
          Returns:
            Tuple of (records list, newest_event_time ISO string)
          """
          api_base = KEEPER_API_BASE.rstrip('/')
          endpoint = f"{api_base}/api/rest/enterprise/audit-events"
      
          headers = {
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
            'User-Agent': 'GoogleSecOps-KeeperCollector/1.0'
          }
      
          records = []
          newest_time = None
          page_num = 0
          backoff = 1.0
      
          # Convert to Unix epoch seconds for Keeper API
          start_epoch = int(start_time.timestamp())
          end_epoch = int(end_time.timestamp())
          cursor = None
      
          while True:
            page_num += 1
      
            if len(records) >= max_records:
              print(f"Reached max_records limit ({max_records})")
              break
      
            # Build request body
            body = {
              'start_time': start_epoch,
              'end_time': end_epoch,
              'limit': min(page_size, max_records - len(records)),
              'enterprise_id': KEEPER_ENTERPRISE_ID
            }
            if cursor:
              body['cursor'] = cursor
      
            try:
              response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body),
                headers=headers
              )
      
              # 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:
                print(f"HTTP Error: {response.status}")
                response_text = response.data.decode('utf-8')
                print(f"Response body: {response_text}")
                return [], None
      
              data = json.loads(response.data.decode('utf-8'))
      
              page_results = data.get('audit_events', data.get('events', []))
      
              if not page_results:
                print(f"No more results (empty page)")
                break
      
              print(f"Page {page_num}: Retrieved {len(page_results)} events")
              records.extend(page_results)
      
              # Track newest event time
              for event in page_results:
                try:
                  event_ts = event.get('timestamp') or event.get('created')
                  if event_ts:
                    if isinstance(event_ts, (int, float)):
                      event_dt = datetime.fromtimestamp(event_ts, tz=timezone.utc)
                      event_time = event_dt.isoformat()
                    else:
                      event_time = str(event_ts)
                    if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                      newest_time = event_time
                except Exception as e:
                  print(f"Warning: Could not parse event time: {e}")
      
              # Check for more results
              cursor = data.get('cursor') or data.get('next_cursor')
              if not cursor:
                print("No more pages (no next cursor)")
                break
      
            except Exception as e:
              print(f"Error fetching logs: {e}")
              return [], None
      
          print(f"Retrieved {len(records)} total records from {page_num} pages")
          return records, newest_time
      
    • 第二個檔案 - 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. 請提供下列設定詳細資料:

    設定
    名稱 keeper-logs-collector-hourly
    區域 選取與 Cloud Run 函式相同的區域
    頻率 0 * * * * (每小時整點)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取主題 keeper-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. 按一下 keeper-logs-collector
  6. 按一下 [Logs] (記錄) 分頁標籤。
  7. 確認函式是否已順利執行。請注意以下事項:

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X events
    Wrote X records to gs://keeper-audit-logs/keeper/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 按一下 bucket 名稱 (keeper-audit-logs)。

  10. 前往 keeper/ 資料夾。

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

如果記錄中顯示錯誤:

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

在 Google SecOps 中設定動態饋給,擷取 Keeper Enterprise Security 記錄

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

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

  9. 點選「下一步」

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

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

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

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

    • 檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 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 對應 邏輯
logData.channel extensions.auth.type 如果值位於 ["SSO", "MACHINE", "VPN", "PHYSICAL", "TACACS"] 中,請設為大寫值,否則請設為「AUTHTYPE_UNSPECIFIED」
logData.timestamp metadata.event_timestamp 以 ISO8601 格式轉換
logData.audit_event、logData.remote_address metadata.event_type 如果 audit_event 符合「login」且 remote_address 不為空白,則設為「USER_LOGIN」;如果 remote_address 不為空白,則設為「STATUS_UPDATE」;否則設為「GENERIC_EVENT」
logData.audit_event metadata.product_event_type 直接複製值
logData.enterprise_id metadata.product_log_id 已轉換為字串
metadata.product_name 設為「KEEPER」
metadata.vendor_name 設為「KEEPER」
logData.client_version network.http.parsed_user_agent 直接複製值,然後轉換為 parseduseragent
logData.client_version network.http.user_agent 直接複製值
logData.remote_address principal.asset.ip 直接複製值
logData.remote_address principal.ip 直接複製值
logData.category security_result.category_details 從來源合併
logData.shared_folder_uid, logData.folder_uid security_result.detection_fields 合併為索引鍵為「shared_folder_uid」和「folder_uid」的標籤
logData.email target.user.email_addresses 從來源合併
logData.username target.user.userid 直接複製值

變更記錄

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

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