收集 1Password 記錄
剖析器版本:5.0
本文說明如何使用 Google Cloud Storage V2,將 1Password 記錄擷取至 Google Security Operations。
1Password 是一項密碼管理平台,可協助團隊安全地儲存、共用及管理憑證、密鑰和機密資訊。1Password Events API 可存取 1Password Business 帳戶的登入嘗試和項目使用資料。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作所需的權限
- 1Password Business 帳戶
- 1Password 帳戶的擁有者或管理員角色
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 onepassword-secops-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或資料保留政策 點選「建立」。
收集 1Password 憑證
設定事件報表整合
- 在 1Password.com 登入帳戶。
- 選取側欄中的「整合」。
- 如果您先前已設定其他整合服務,請在「整合」頁面選取「目錄」。
- 在「事件回報功能」部分,選取「其他」。
- 在「Name」欄位中,輸入整合的名稱 (例如
Google SecOps Integration)。 - 按一下「新增整合」。
建立不記名權杖
- 在整合詳細資料頁面中,按一下「新增不記名權杖」。
- 請提供下列設定詳細資料:
- 權杖名稱:輸入描述性名稱 (例如
SecOps GCS Collector - SignIn and ItemUsage) - 到期時間:根據偏好設定選取有效期限。如要建立永不過期的權杖,請選取「永不過期」,或選擇「30 天」、「90 天」或「180 天」。
- 要回報的事件:勾選下列核取方塊:
- 嘗試登入
- 項目用量事件
- 權杖名稱:輸入描述性名稱 (例如
- 按一下「Issue Token」。
在「Save your token」頁面,按一下「Save in 1Password」,或複製權杖並儲存在安全的位置。
按一下「查看整合詳細資料」,確認整合功能已啟用。
判斷 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 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
onepassword-logs-collector-sa - 服務帳戶說明:輸入
Service account for Cloud Run function to collect 1Password logs
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run 叫用者」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions 叫用者」。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS bucket,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許叫用函式
授予 GCS bucket 的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶:
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket (
onepassword-secops-logs)。 - 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入
onepassword-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,供 Cloud Scheduler 發布訊息,以及 Cloud Run 函式訂閱訊息。
- 在 GCP 控制台中,前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
onepassword-logs-trigger - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 1Password Events API 擷取登入嘗試和項目使用記錄,並寫入 GCS。
- 在 GCP 控制台中,前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 onepassword-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件 (選用)」部分:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇
onepassword-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁標籤:
- 「服務帳戶」:選取
onepassword-logs-collector-sa
- 「服務帳戶」:選取
前往「容器」分頁:
- 按一下「Variables & Secrets」(變數與密鑰)。
針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETonepassword-secops-logsGCS bucket 名稱 GCS_PREFIXonepassword記錄檔的前置字串 STATE_KEYonepassword/state.json狀態檔案路徑 OP_API_BASEhttps://events.1password.com1Password Events API 基準網址 OP_BEARER_TOKEN<your-bearer-token>1Password Events API 不記名權杖 MAX_RECORDS10000每次執行時,每個端點的記錄數量上限 PAGE_SIZE1000每頁記錄數 (最多 1000 筆) LOOKBACK_HOURS24初始回溯期 (以小時為單位)
在「變數與密鑰」部分,捲動至「要求」:
- 要求逾時:輸入
600秒 (10 分鐘)
- 要求逾時:輸入
前往「設定」分頁:
- 在「資源」部分:
- 記憶體:選取 512 MiB 以上
- CPU:選取 1
- 在「資源」部分:
在「修訂版本資源調度」部分:
- 執行個體數量下限:輸入
0 - 執行個體數量上限:輸入
100(或根據預期負載調整)
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「進入點」欄位中輸入 main。
在內嵌程式碼編輯器中,建立兩個檔案:
第一個檔案: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', } 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
點選「部署」即可儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題,觸發 Cloud Run 函式。
- 前往 GCP 控制台中的「Cloud Scheduler」。
- 點選「建立工作」。
提供下列設定詳細資料:
設定 值 名稱 onepassword-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議選取世界標準時間) 目標類型 Pub/Sub 主題 選取「 onepassword-logs-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
請根據記錄量和延遲時間需求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
高容量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等音量 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
低量、批次處理 |
| 每天 | 0 0 * * * |
歷來資料集合 |
測試整合
- 在 Cloud Scheduler 控制台中,找到
onepassword-logs-collector-hourly。 - 按一下「強制執行」,手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「服務」。
- 按一下「
onepassword-logs-collector」。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。尋找:
--- 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依序前往「Cloud Storage」>「Buckets」。
按一下「
onepassword-secops-logs」。前往
onepassword/資料夾。確認已建立新的
.ndjson檔案,且時間戳記為目前時間。
如果記錄中顯示錯誤:
- HTTP 401:檢查
OP_BEARER_TOKEN環境變數中的不記名權杖。權杖可能已過期。 - HTTP 429:速率限制。函式會自動重試,並進行輪詢。1Password Events API 每分鐘可發送 600 項要求,每小時可發送 30,000 項要求。
- 缺少環境變數:檢查 Cloud Run 函式設定中是否已設定所有必要變數。
- 未傳回任何記錄:使用內省端點,確認不記名權杖可存取要求的事件類型。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中,輸入動態饋給名稱 (例如
1Password Sign-In and Item Usage Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「1Password」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket 網址:輸入 GCS bucket URI,並加上前置路徑:
gs://onepassword-secops-logs/onepassword/
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:一律不會在轉移後刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移檔案後,刪除檔案和空白目錄。
檔案存在時間上限:納入在過去天數內修改的檔案 (預設為 180 天)
資產命名空間:資產命名空間
擷取標籤:要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶必須具備 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下「
onepassword-secops-logs」。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色:選取「Storage 物件檢視者」
按一下 [儲存]。
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 平台值 (LINUX、WINDOWS、MAC)。 |
| 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 | 如果類別為 success 或 firewall_reported_success,請設為 USER_LOGIN;如果沒有使用者資訊,請設為 STATUS_UPDATE;否則請設為 USER_UNCATEGORIZED。 |
| 不適用 | metadata.log_type | 設為 ONEPASSWORD。 |
| 不適用 | metadata.product_name | 設為 ONEPASSWORD。 |
| 不適用 | metadata.vendor_name | 設為 ONEPASSWORD。 |
| 不適用 | security_result.action | 如果類別為 success 或 firewall_reported_success,請設為 ALLOW;如果類別為 credentials_failed、mfa_failed、modern_version_failed 或 firewall_failed,請設為 BLOCK;否則請留空。 |
lt |
event.idm.read_only_udm.metadata.product_event_type |
從變更記錄對應 |
item.actor_details.uuid |
principal.user.userid |
從變更記錄對應 |
item.actor_details.name |
principal.user.user_display_name |
從變更記錄對應 |
item.actor_details.email |
principal.email |
從變更記錄對應 |
item.actor_uuid |
security_result.about.resource.attribute.labels |
從變更記錄對應 |
item.object_type |
additional.fields |
從變更記錄對應 |
item.object_uuid |
security_result.about.resource.attribute.labels |
從變更記錄對應 |
item.aux_info |
additional.fields |
從變更記錄對應 |
item.aux_uuid |
security_result.about.resource.attribute.labels |
從變更記錄對應 |
item.session.uuid |
network.session_id |
從變更記錄對應 |
item.session.device_uuid |
target.resource.product_object_id |
從變更記錄對應 |
item.session.ip |
principal.ip |
從變更記錄對應 |
target_user.name |
target.user.user_display_name |
從變更記錄對應 |
uuid |
principal.userid |
從變更記錄對應 |
type","action","used_version", |
additional.fields |
從變更記錄對應 |
target_user.uuid |
target.user.product_object_id |
從變更記錄對應 |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求解答。