收集 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 值區
- 前往 Google Cloud Console。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「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」。
在「儲存權杖」頁面中,按一下「儲存至 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 & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
onepassword-logs-collector-sa - 服務帳戶說明:輸入
Service account for Cloud Run function to collect 1Password logs
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket (
onepassword-secops-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入
onepassword-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
onepassword-logs-trigger - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 1Password Events API 擷取登入嘗試和項目使用記錄,並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 onepassword-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中,選擇
onepassword-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁:
- 服務帳戶:選取
onepassword-logs-collector-sa
- 服務帳戶:選取
前往「容器」分頁:
- 按一下「變數與密鑰」。
針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 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每頁記錄數 (最多 1,000 筆) LOOKBACK_HOURS24初始回溯期 (以小時為單位)
在「變數與 Secret」部分,捲動至「要求」:
- 要求逾時:輸入
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', '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
點選「部署」來儲存並部署函式。
等待部署作業完成 (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」>「Services」。
- 按一下「
onepassword-logs-collector」。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找:
--- 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 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
1Password Sign-In and Item Usage Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「1Password」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示不重複的服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 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;否則請留空。 |
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。