收集 1Password 稽核事件記錄

剖析器版本:2.0

支援語言:

本文說明如何使用 Cloud Storage V2,將 1Password 稽核事件記錄檔擷取至 Google Security Operations。

1Password 是一個密碼管理平台,可協助團隊安全地儲存、共用及管理憑證、密鑰和機密資訊。透過 1Password Events API,您可以存取稽核事件資料,瞭解 1Password Business 帳戶中執行的管理和政策動作。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 Google Cloud 專案
  • 建立及管理 Cloud Storage bucket 的權限
  • 管理 Cloud Storage 值區身分與存取權管理 (IAM) 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • 1Password Business 帳戶
  • 1Password 帳戶的擁有者或管理員角色

建立 Cloud Storage bucket

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

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

收集 1Password 憑證

設定事件報表整合

  1. 1Password.com 登入帳戶。
  2. 選取側欄中的「整合」
  3. 如果您先前已設定其他整合服務,請在「整合」頁面選取「目錄」
  4. 在「事件回報功能」部分,選取「其他」
  5. 在「Name」欄位中,輸入整合的名稱 (例如 Google SecOps Audit Events)。
  6. 按一下「新增整合」

建立不記名權杖

  1. 1Password 會直接進入不記名權杖設定程序。

  2. 請提供下列設定詳細資料:

    • 權杖名稱:輸入描述性名稱 (例如 SecOps GCS Collector - Audit Events)
    • 有效期限:選擇權杖的有效期限,包括 30 天90 天180 天。預設值為「永不」
    • 要回報的事件:選擇權杖可存取的事件。系統預設會選取所有事件類型。選取「稽核事件」,並取消選取「登入嘗試」和「項目使用情形」,確保權杖僅限稽核事件。
  3. 按一下「核發權杖」

  4. 在「Save your token」(儲存權杖) 頁面中,按一下「Save in 1Password」(儲存至 1Password),選擇要儲存權杖的保險箱,然後按一下「Save」(儲存)。權杖會儲存為 API 憑證項目。

  5. 按一下「查看整合詳細資料」,確認整合項目已啟用。

判斷 Events API 基本網址

基本網址取決於代管 1Password 帳戶的伺服器:

如果帳戶代管於 Events API 的基本網址為
1password.com https://events.1password.com
ent.1password.com https://events.ent.1password.com
1password.ca https://events.1password.ca
1password.eu https://events.1password.eu

測試 API 存取權

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

    # Replace with your actual bearer token and base URL
    BEARER_TOKEN="<your-bearer-token>"
    API_BASE="https://events.1password.com"
    
    # Test API access using the introspect endpoint
    curl -v \
        -H "Authorization: Bearer $BEARER_TOKEN" \
        "$API_BASE/api/v2/auth/introspect"
    

如果回應成功,系統會傳回 JSON 物件,其中 features 欄位會列出權杖可存取的事件類型 (例如 ["auditevents"])。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於:

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

授予 Cloud Storage bucket 的 IAM 權限

授予服務帳戶 Cloud Storage 值區的寫入權限:

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 按一下 bucket (onepassword-audit-secops-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 提供下列設定詳細資料:
    • 新增主體:輸入 onepassword-audit-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:輸入 onepassword-audit-trigger
    • 其他設定保留預設值
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 1Password Events API 擷取稽核事件記錄,並將其寫入 Cloud Storage。

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

    設定
    服務名稱 onepassword-audit-collector
    區域 選取與 Cloud Storage bucket 相符的區域 (例如 us-central1)
    執行階段 選取 Python 3.12 以上版本
  5. 在「觸發條件 (選用)」部分:

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

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

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

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

    1. 按一下「變數與密鑰」
    2. 針對每個環境變數,按一下「+ 新增變數」
    變數名稱 範例值 說明
    GCS_BUCKET onepassword-audit-secops-logs Cloud Storage Bucket 名稱
    GCS_PREFIX onepassword-audit 記錄檔的前置字串
    STATE_KEY onepassword-audit-state.json 狀態檔案路徑,位於記錄前置字串外部
    OP_API_BASE https://events.1password.com 1Password Events API 基準網址
    OP_BEARER_TOKEN <your-bearer-token> 1Password Events API 不記名權杖
    MAX_RECORDS 10000 每次執行的記錄數上限
    PAGE_SIZE 1000 每頁記錄數 (最多 1,000 筆)
    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
      from google.cloud.exceptions import NotFound
      import json
      import os
      import re
      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', 'onepassword-audit')
      # STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
      # its bucket URI and, with a deletion option selected, deletes what it
      # transferred. A state file inside the prefix is ingested as log data and then
      # deleted, which drops the cursor and replays the whole lookback window.
      STATE_KEY = os.environ.get('STATE_KEY', 'onepassword-audit-state.json')
      API_BASE = os.environ.get('OP_API_BASE')
      BEARER_TOKEN = os.environ.get('OP_BEARER_TOKEN')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
      # The Events API accepts a limit from 1 to 1000.
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      # 1Password Events API v2 audit events endpoint
      AUDIT_EVENTS_PATH = '/api/v2/auditevents'
      
      class FetchError(Exception):
          """Raised when the 1Password Events API call fails.
      
          State must never be written on a failed fetch. Saving a rejected cursor back
          would wedge the collector: every later run would fail on page 1, ingest
          nothing, and still report success.
          """
      
      def parse_datetime(value: str) -> datetime:
          """Parse a 1Password `timestamp` into an aware datetime.
      
          The API returns RFC 3339 with either a numeric offset and whole seconds
          (2023-03-15T16:33:50-03:00) or Z with nanosecond precision
          (2025-10-31T13:45:49.203617068Z). Fractional digits are trimmed to
          microseconds because fromisoformat accepts at most six.
          """
          text = str(value).strip()
          if text.endswith('Z'):
              text = text[:-1] + '+00:00'
          text = re.sub(r'\.(\d{6})\d+', r'.\1', text)
          dt = datetime.fromisoformat(text)
          if dt.tzinfo is None:
              dt = dt.replace(tzinfo=timezone.utc)
          return dt.astimezone(timezone.utc)
      
      @functions_framework.cloud_event
      def main(cloud_event):
          """Fetch 1Password audit events and write them to Cloud Storage as NDJSON."""
          if not all([GCS_BUCKET, API_BASE, BEARER_TOKEN]):
              # Raise rather than return: a bare return acks the Pub/Sub message and
              # reports the run as successful, silently discarding the schedule tick.
              raise RuntimeError('Missing required environment variables '
                              '(GCS_BUCKET, OP_API_BASE, OP_BEARER_TOKEN)')
      
          bucket = storage_client.bucket(GCS_BUCKET)
          state = load_state(bucket, STATE_KEY)
      
          now = datetime.now(timezone.utc)
      
          print('--- Processing endpoint: auditevents ---')
      
          saved_cursor = state.get('cursor_auditevents')
          seen_uuids = set(state.get('seen_uuids', []))
      
          # If the cursor is gone, resume from the last event actually written instead
          # of replaying a blind 24 hours. LOOKBACK_HOURS applies only on a cold start.
          if saved_cursor:
              start_time = None
          elif state.get('last_event_time'):
              start_time = parse_datetime(state['last_event_time'])
              print(f'No cursor found. Resuming from the last recorded event at {start_time.isoformat()}')
          else:
              start_time = now - timedelta(hours=LOOKBACK_HOURS)
              print(f'Cold start. Fetching from {start_time.isoformat()}')
      
          records, last_cursor = fetch_endpoint(
              api_base=API_BASE,
              path=AUDIT_EVENTS_PATH,
              bearer_token=BEARER_TOKEN,
              saved_cursor=saved_cursor,
              start_time=start_time,
              page_size=PAGE_SIZE,
              max_records=MAX_RECORDS,
          )
      
          # The cursor path can re-deliver events, so filter by uuid before writing.
          fresh = [r for r in records if str(r.get('uuid', '')) not in seen_uuids]
          print(f'Fetched {len(records)} records, {len(fresh)} new after deduplication')
      
          if fresh:
              timestamp = now.strftime('%Y%m%dT%H%M%SZ')
              object_key = f'{GCS_PREFIX}/auditevents_{timestamp}.ndjson'
              blob = bucket.blob(object_key)
      
              ndjson = '\n'.join(
                  json.dumps(r, ensure_ascii=False) for r in fresh
              ) + '\n'
              blob.upload_from_string(
                  ndjson, content_type='application/x-ndjson'
              )
      
              print(f'Wrote {len(fresh)} auditevents records '
                  f'to gs://{GCS_BUCKET}/{object_key}')
      
          # State is written only after any data is durably uploaded.
          new_state = dict(state)
          if last_cursor:
              new_state['cursor_auditevents'] = last_cursor
      
          newest = None
          for record in records:
              stamp = record.get('timestamp')
              if not stamp:
                  continue
              try:
                  parsed = parse_datetime(stamp)
              except ValueError:
                  continue
              if newest is None or parsed > newest:
                  newest = parsed
          if newest:
              new_state['last_event_time'] = newest.isoformat()
              # Retain the uuids of the newest batch so a cursor reset does not
              # re-ingest the events that straddle the resume point.
              new_state['seen_uuids'] = sorted(
                  {str(r.get('uuid', '')) for r in records if r.get('uuid')}
              )
      
          new_state['last_run'] = now.isoformat()
          save_state(bucket, STATE_KEY, new_state)
      
          print(f'Successfully processed {len(fresh)} records')
      
      def load_state(bucket, key):
          """Read the collector state from Cloud Storage.
      
          Only a missing object is treated as a cold start. Any other error is raised:
          swallowing it would drop the cursor and replay the lookback window.
          """
          blob = bucket.blob(key)
          try:
              return json.loads(blob.download_as_text())
          except NotFound:
              print('No state file found. Starting from the lookback window.')
              return {}
      
      def save_state(bucket, key, state: dict):
          """Write the collector state to Cloud Storage.
      
          Failures are raised, not logged. If the state write fails after the data was
          uploaded, the next run resumes from the stale cursor and duplicates it.
          """
          blob = bucket.blob(key)
          blob.upload_from_string(
              json.dumps(state, indent=2),
              content_type='application/json',
          )
          print(f"Saved state: cursor={state.get('cursor_auditevents')!r} "
              f"last_event_time={state.get('last_event_time')}")
      
      def fetch_endpoint(api_base, path, bearer_token, saved_cursor,
                      start_time, page_size, max_records):
          """Fetch events from the 1Password Events API v2 audit events endpoint.
      
          The API uses cursor-based pagination with POST requests. The first request
          sends a ResetCursor object with an optional start_time and limit. Subsequent
          requests send the cursor returned by the previous response. The cursor is a
          persistent checkpoint and stays valid across runs.
      
          Args:
              api_base: Events API base URL
              path: Endpoint path
              bearer_token: JWT bearer token
              saved_cursor: Cursor from the previous run, or None
              start_time: Lower bound used only when there is no cursor
              page_size: Max events per page (1 to 1000)
              max_records: Max total events per run
      
          Returns:
              Tuple of (records list, cursor to persist).
      
          Raises:
              FetchError: on any API or transport failure, so that a rejected cursor is
                  never written back to the state file.
          """
          url = f'{api_base.rstrip("/")}{path}'
      
          headers = {
              'Authorization': f'Bearer {bearer_token}',
              'Content-Type': 'application/json',
              'Accept': 'application/json',
          }
      
          records = []
          cursor = saved_cursor
          page_num = 0
          backoff = 1.0
          rate_limit_retries = 0
          MAX_RATE_LIMIT_RETRIES = 5
      
          while True:
              page_num += 1
      
              if len(records) >= max_records:
                  print(f'Reached max_records limit ({max_records})')
                  break
      
              if cursor:
                  # Continuing cursor: resume from the last position.
                  body = json.dumps({'cursor': cursor})
              else:
                  # ResetCursor: first request, or the cursor was lost.
                  reset = {'limit': page_size}
                  if start_time:
                      reset['start_time'] = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
                  body = json.dumps(reset)
      
              try:
                  response = http.request(
                      'POST', url, body=body, headers=headers,
                  )
              except Exception as e:
                  raise FetchError(f'Request to {url} failed: {e}') from e
      
              # Documented limits are 600 requests per minute and 30,000 per hour.
              if response.status == 429:
                  rate_limit_retries += 1
                  if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                      raise FetchError('Rate limited repeatedly; giving up without saving state')
                  raw_retry_after = response.headers.get('Retry-After')
                  try:
                      retry_after = int(raw_retry_after) if raw_retry_after else int(backoff)
                  except (TypeError, ValueError):
                      retry_after = int(backoff)
                  print(f'Rate limited (429). Retrying after {retry_after}s...')
                  time.sleep(retry_after)
                  backoff = min(backoff * 2, 60.0)
                  continue
      
              backoff = 1.0
              rate_limit_retries = 0
      
              if response.status != 200:
                  body_text = response.data.decode('utf-8')
                  # A 400 or 401 here may mean the saved cursor was rejected. Raising
                  # leaves the previous state intact so the run can be retried, rather
                  # than persisting a cursor that stalls every future run.
                  raise FetchError(f'HTTP {response.status} from the Events API: {body_text}')
      
              try:
                  data = json.loads(response.data.decode('utf-8'))
              except json.JSONDecodeError as e:
                  raise FetchError(f'Malformed JSON response from the Events API: {e}') from e
      
              page_items = data.get('items', [])
              cursor = data.get('cursor')
              has_more = data.get('has_more', False)
      
              if page_items:
                  print(f'Page {page_num}: Retrieved '
                      f'{len(page_items)} events')
                  records.extend(page_items)
      
              if not has_more:
                  print('No more pages (has_more=false)')
                  break
      
              if not cursor:
                  print('No cursor returned, stopping')
                  break
      
              if not page_items:
                  # has_more can be true while a page is empty. Stop rather than spin.
                  print('Empty page with has_more=true; resuming on the next run')
                  break
      
          print(f'Retrieved {len(records)} total records '
              f'from {page_num} pages')
          return records, cursor
      

    • 第二個檔案 - 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. 請提供下列設定詳細資料:

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

排程頻率選項

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

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

測試整合項目

  1. Cloud Scheduler 控制台中,找出 onepassword-audit-collector-hourly
  2. 按一下「強制執行」即可手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「Services」
  5. 按一下「onepassword-audit-collector」。
  6. 點選「記錄」分頁標籤。
  7. 確認函式是否執行成功。尋找:

    --- Processing endpoint: auditevents ---
    Page 1: Retrieved X events
    Fetched X records, Y new after deduplication
    Wrote Y auditevents records to gs://onepassword-audit-secops-logs/onepassword-audit/auditevents_YYYYMMDDTHHMMSSZ.ndjson
    Saved state: cursor='...' last_event_time=YYYY-MM-DDTHH:MM:SS+00:00
    Successfully processed Y records
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 按一下「onepassword-audit-secops-logs」。

  10. 前往 onepassword-audit/ 資料夾。

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

如果記錄中顯示錯誤:

  • HTTP 401:檢查 OP_BEARER_TOKEN 環境變數中的不記名權杖。權杖可能已過期。
  • HTTP 429:速率限制。函式會自動重試,並進行輪詢。1Password Events API 每分鐘可發送 600 項要求,每小時可發送 30,000 項要求。
  • 缺少環境變數:檢查 Cloud Run 函式設定中是否已設定所有必要變數。
  • 未傳回任何記錄:使用內省端點,確認持有人權杖可存取稽核事件。
  • 重複收到相同事件:確認 STATE_KEY 不在動態饋給讀取的字首內。如果動態饋給刪除狀態檔案,游標就會遺失,收集器會從最後記錄的事件重新讀取。

擷取 Google SecOps 服務帳戶

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

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

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

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

  10. 點選「下一步」

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

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

      gs://onepassword-audit-secops-logs/onepassword-audit/
      
    • 來源刪除選項:根據偏好設定選取刪除選項:

      • 一律不刪除檔案:一律不刪除來源中的檔案 (建議用於測試)。
      • 刪除已轉移的檔案和空白目錄:成功擷取檔案後,從來源刪除檔案和空白目錄。

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

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

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

  12. 點選「下一步」

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

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

Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩個角色:Storage 物件檢視者,用於讀取記錄物件;以及 bucket 層級角色,用於讀取 bucket 中繼資料。

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 按一下「onepassword-audit-secops-logs」。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 提供下列設定詳細資料:
    • 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
    • 指派角色:選取下列兩個角色:
      • Storage 物件檢視者:讀取記錄檔物件。
      • Storage 舊版值區讀取者:讀取值區中繼資料。如果您選取「刪除已轉移的檔案和空目錄」刪除選項,請改為選取「Storage Legacy Bucket Writer」,這也會授予刪除權限。
  6. 按一下 [儲存]

UDM 對應表

記錄欄位 UDM 對應 邏輯
additional_action additional.fields 已合併
aux_id_label additional.fields 已合併
aux_info_list additional.fields 已合併
object_type_label additional.fields 已合併
EventTime metadata.event_timestamp 已剖析為 yyyy-MM-dd HH:mm:ss
expiry metadata.event_timestamp 已剖析為 yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ
item.session.login_time metadata.event_timestamp 已剖析為 yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ
item.timestamp metadata.event_timestamp 已剖析為 ISO8601
has_principal metadata.event_type 已對應:trueUSER_UNCATEGORIZEDtrueSTATUS_UPDATE
item.session.uuid network.session_id 直接對應
deviceUuid principal.asset.asset_id 直接對應
session_ip principal.asset.ip 已合併
item.actor_details.email principal.email 直接對應
session_ip principal.ip 已合併
item.location.city principal.location.city 直接對應
item.location.country principal.location.country_or_region 直接對應
item.location.region principal.location.name 直接對應
item.location.latitude principal.location.region_latitude 直接對應
item.location.longitude principal.location.region_longitude 直接對應
uuid_label principal.resource.attribute.labels 已合併
item.actor_details.name principal.user.user_display_name 直接對應
item.actor_details.uuid principal.user.userid 直接對應
actor_uuid_label security_result.about.resource.attribute.labels 已合併
aux_uid_label security_result.about.resource.attribute.labels 已合併
object_id_label security_result.about.resource.attribute.labels 已合併
item.session.device_uuid target.resource.product_object_id 直接對應
不適用 metadata.event_type 常數:USER_UNCATEGORIZED
不適用 metadata.product_name 常數:ONEPASSWORD_AUDIT_EVENTS
不適用 metadata.vendor_name 常數:ONEPASSWORD

變更記錄

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

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