收集 1Password 記錄

支援的國家/地區:

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

1Password 是一個密碼管理平台,可協助團隊安全地儲存、共用及管理憑證、密鑰和機密資訊。1Password Events API 可讓您存取 1Password Business 帳戶的登入嘗試和項目使用資料。

事前準備

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

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

建立 Google Cloud Storage 值區

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

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

收集 1Password 憑證

設定事件報表整合

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

建立不記名權杖

  1. 在整合詳細資料頁面中,按一下「新增不記名權杖」
  2. 請提供下列設定詳細資料:
    • 權杖名稱:輸入描述性名稱 (例如 SecOps GCS Collector - SignIn and ItemUsage)
    • 到期時間:根據偏好設定選取有效期限。如要建立永不過期的權杖,請選取「永不過期」,或選擇「30 天」、「90 天」或「180 天」
    • 要回報的事件:勾選下列核取方塊:
      • 嘗試登入次數
      • 項目使用事件
  3. 按一下「Issue Token」
  4. 在「儲存權杖」頁面中,按一下「儲存至 1Password」,或複製權杖並儲存在安全的位置。

  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 欄位會列出權杖可存取的事件類型 (例如 ["signinattempts", "itemusages"])。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 1Password Events API 擷取登入嘗試和項目使用記錄,並寫入 GCS。

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

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

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

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

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

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

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

      變數名稱 範例值 說明
      GCS_BUCKET onepassword-secops-logs GCS bucket 名稱
      GCS_PREFIX onepassword 記錄檔的前置字串
      STATE_KEY onepassword/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. 在「變數與 Secret」部分,捲動至「要求」

    • 要求逾時:輸入 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
      
      # 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')
      STATE_KEY = os.environ.get('STATE_KEY', 'onepassword/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'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      # 1Password Events API v2 endpoints for sign-in and item usage
      ENDPOINTS = {
        'signinattempts': '/api/v2/signinattempts',
        'itemusages': '/api/v2/itemusages',
      }
      
      def parse_datetime(value: str) -> datetime:
        """Parse RFC 3339 datetime string to datetime object."""
        if value.endswith('Z'):
          value = value[:-1] + '+00:00'
        return datetime.fromisoformat(value)
      
      @functions_framework.cloud_event
      def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch 1Password
        sign-in attempt and item usage logs and write them to GCS
        as NDJSON.
        """
        if not all([GCS_BUCKET, API_BASE, BEARER_TOKEN]):
          print('Error: Missing required environment variables '
              '(GCS_BUCKET, OP_API_BASE, OP_BEARER_TOKEN)')
          return
      
        try:
          bucket = storage_client.bucket(GCS_BUCKET)
      
          # Load state (stores cursors per endpoint)
          state = load_state(bucket, STATE_KEY)
      
          now = datetime.now(timezone.utc)
          total_records = 0
      
          for event_type, path in ENDPOINTS.items():
            print(f'--- Processing endpoint: {event_type} ---')
      
            saved_cursor = state.get(f'cursor_{event_type}')
      
            records, last_cursor = fetch_endpoint(
              api_base=API_BASE,
              path=path,
              bearer_token=BEARER_TOKEN,
              saved_cursor=saved_cursor,
              lookback_hours=LOOKBACK_HOURS,
              page_size=PAGE_SIZE,
              max_records=MAX_RECORDS,
            )
      
            if records:
              timestamp = now.strftime('%Y%m%d_%H%M%S')
              object_key = (f'{GCS_PREFIX}/'
                      f'{event_type}_{timestamp}.ndjson')
              blob = bucket.blob(object_key)
      
              ndjson = '\n'.join(
                json.dumps(r, ensure_ascii=False) for r in records
              ) + '\n'
              blob.upload_from_string(
                ndjson, content_type='application/x-ndjson'
              )
      
              print(f'Wrote {len(records)} {event_type} records '
                  f'to gs://{GCS_BUCKET}/{object_key}')
      
            # Save cursor even if no records (for next poll)
            if last_cursor:
              state[f'cursor_{event_type}'] = last_cursor
      
            total_records += len(records)
      
          # Save state with all cursors
          state['last_run'] = now.isoformat()
          save_state(bucket, STATE_KEY, state)
      
          print(f'Successfully processed {total_records} total 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, 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: {json.dumps(state)}')
        except Exception as e:
          print(f'Warning: Could not save state: {e}')
      
      def fetch_endpoint(api_base, path, bearer_token, saved_cursor,
                lookback_hours, page_size, max_records):
        """
        Fetch events from a single 1Password Events API v2 endpoint.
      
        The 1Password Events API uses cursor-based pagination with POST
        requests. The first request sends a ResetCursor object with
        optional start_time and limit. Subsequent requests send the
        cursor returned from the previous response.
      
        Args:
          api_base: Events API base URL
          path: Endpoint path
          bearer_token: JWT bearer token
          saved_cursor: Cursor from previous run (or None)
          lookback_hours: Hours to look back on first run
          page_size: Max events per page (1-1000)
          max_records: Max total events per run
      
        Returns:
          Tuple of (records list, last_cursor string)
        """
        url = f'{api_base.rstrip("/")}{path}'
      
        headers = {
          'Authorization': f'Bearer {bearer_token}',
          'Content-Type': 'application/json',
          'Accept': 'application/json',
          'User-Agent': 'GoogleSecOps-1PasswordCollector/1.0',
        }
      
        records = []
        cursor = saved_cursor
        page_num = 0
        backoff = 1.0
      
        while True:
          page_num += 1
      
          if len(records) >= max_records:
            print(f'Reached max_records limit ({max_records})')
            break
      
          # Build request body
          if cursor:
            # Continuing cursor: resume from last position
            body = json.dumps({'cursor': cursor})
          else:
            # ResetCursor: first request or no saved state
            start_time = (
              datetime.now(timezone.utc)
              - timedelta(hours=lookback_hours)
            )
            body = json.dumps({
              'limit': page_size,
              'start_time': start_time.strftime(
                '%Y-%m-%dT%H:%M:%SZ'
              ),
            })
      
          try:
            response = http.request(
              'POST', url, body=body, headers=headers,
            )
      
            # Handle rate limiting (600 req/min, 30000 req/hour)
            if response.status == 429:
              retry_after = int(
                response.headers.get(
                  'Retry-After', str(int(backoff))
                )
              )
              print(f'Rate limited (429). Retrying after '
                  f'{retry_after}s...')
              time.sleep(retry_after)
              backoff = min(backoff * 2, 60.0)
              continue
      
            backoff = 1.0
      
            if response.status != 200:
              response_text = response.data.decode('utf-8')
              print(f'HTTP Error {response.status}: '
                  f'{response_text}')
              break
      
            data = json.loads(response.data.decode('utf-8'))
      
            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(f'No more pages (has_more=false)')
              break
      
            if not cursor:
              print(f'No cursor returned, stopping')
              break
      
          except urllib3.exceptions.HTTPError as e:
            print(f'HTTP error: {str(e)}')
            break
          except Exception as e:
            print(f'Error fetching events: {str(e)}')
            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-logs-collector-hourly
    區域 選取與 Cloud Run 函式相同的區域
    頻率 0 * * * * (每小時整點)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取「onepassword-logs-trigger
    郵件內文 {} (空白 JSON 物件)
  4. 點選「建立」

排程頻率選項

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

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

測試整合項目

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

    --- Processing endpoint: signinattempts ---
    Page 1: Retrieved X events
    Wrote X signinattempts records to gs://onepassword-secops-logs/onepassword/signinattempts_YYYYMMDD_HHMMSS.ndjson
    --- Processing endpoint: itemusages ---
    Page 1: Retrieved X events
    Wrote X itemusages records to gs://onepassword-secops-logs/onepassword/itemusages_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X total records
    
  8. 依序前往「Cloud Storage」>「Buckets」

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

  10. 前往 onepassword/ 資料夾。

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

如果在記錄中發現錯誤:

  • HTTP 401:檢查 OP_BEARER_TOKEN 環境變數中的不記名權杖。權杖可能已過期。
  • HTTP 429:速率限制。函式會自動重試,並進行輪詢。1Password Events API 每分鐘可發送 600 項要求,每小時可發送 30,000 項要求。
  • 缺少環境變數:檢查 Cloud Run 函式設定中是否已設定所有必要變數。
  • 未傳回任何記錄:使用內省端點,確認持有人權杖可存取要求的事件類型。

擷取 Google SecOps 服務帳戶

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

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

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

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

  9. 點選「下一步」

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

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

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

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

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

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

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

  11. 點選「下一步」

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

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

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

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

UDM 對應表

記錄欄位 UDM 對應 邏輯
category security_result.category_details 這個值取自原始記錄中的 category 欄位。
client.app_name principal.application 這個值取自原始記錄中的 client.app_name 欄位。
client.app_version metadata.product_version 這個值取自原始記錄中的 client.app_version 欄位。
client.ip_address principal.ip 這個值取自原始記錄中的 client.ip_address 欄位。
client.os_name principal.platform 這個值取自原始記錄中的 client.os_name 欄位,並對應至相應的 UDM 平台值 (LINUXWINDOWSMAC)。
client.os_version principal.platform_version 這個值取自原始記錄中的 client.os_version 欄位。
client.platform_name principal.resource.attribute.labels 這個值取自原始記錄中的 client.platform_name 欄位。
client.platform_version principal.asset.platform_software.platform_version 這個值取自原始記錄中的 client.platform_version 欄位。
國家/地區 principal.location.country_or_region 如果沒有 location.country,系統會從原始記錄的 country 欄位取得值。
item_uuid security_result.about.resource.attribute.labels 這個值取自原始記錄中的 item_uuid 欄位。
location.city principal.location.city 這個值取自原始記錄中的 location.city 欄位。
location.country principal.location.country_or_region 這個值取自原始記錄中的 location.country 欄位。
location.latitude principal.location.region_latitude 這個值取自原始記錄中的 location.latitude 欄位。
location.longitude principal.location.region_longitude 這個值取自原始記錄中的 location.longitude 欄位。
location.region principal.location.name 這個值取自原始記錄中的 location.region 欄位。
session.ip principal.ip 這個值取自原始記錄中的 session.ip 欄位。
session_uuid network.session_id 這個值取自原始記錄中的 session_uuid 欄位。
target_user.email target.user.email_addresses 這個值取自原始記錄中的 target_user.email 欄位。
target_user.uuid target.user.userid 這個值取自原始記錄中的 target_user.uuid 欄位。
時間戳記 metadata.event_timestamp 這個值取自原始記錄中的 timestamp 欄位,並轉換為秒和奈秒。
類型 additional.fields 這個值取自原始記錄中的 type 欄位。
user.email principal.user.email_addresses 這個值取自原始記錄中的 user.email 欄位。
user.name principal.user.user_display_name 這個值取自原始記錄中的 user.name 欄位。
used_version additional.fields 這個值取自原始記錄中的 used_version 欄位。
uuid principal.resource.attribute.labels 這個值取自原始記錄中的 uuid 欄位。
vault_uuid security_result.about.resource.attribute.labels 這個值取自原始記錄中的 vault_uuid 欄位。
不適用 extensions.auth 系統會為這個欄位建立空白物件。
不適用 metadata.event_type 如果類別為 successfirewall_reported_success,請設為 USER_LOGIN;如果沒有使用者資訊,請設為 STATUS_UPDATE;否則請設為 USER_UNCATEGORIZED
不適用 metadata.log_type 設為 ONEPASSWORD
不適用 metadata.product_name 設為 ONEPASSWORD
不適用 metadata.vendor_name 設為 ONEPASSWORD
不適用 security_result.action 如果類別為 successfirewall_reported_success,請設為 ALLOW;如果類別為 credentials_failedmfa_failedmodern_version_failedfirewall_failed,請設為 BLOCK;否則請留空。

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