收集 Entrust 身分驗證 (原名 Onfido) 記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage V2，將 Entrust 身分驗證 (舊稱 Onfido) 記錄擷取至 Google Security Operations。剖析器會將原始的 Onfido 驗證檢查和報告記錄，轉換為 Google SecOps UDM 結構定義。
Entrust Identity Verification 是雲端式身分驗證平台，可自動執行文件驗證、生物辨識技術分析和詐欺偵測。這項服務提供 REST API，可用於管理申請人、驗證檢查和報告，讓機構將身分驗證工作流程整合至應用程式。
事前準備
請確認您符合下列必要條件：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作所需的權限
- 具備 API 存取權的 Entrust 身分驗證 (原稱 Onfido) 帳戶
- 具備足夠權限的 Onfido API 權杖，可讀取檢查和報告
收集 Onfido API 憑證
取得 API 權杖
- 登入 Onfido 資訊主頁。
- 依序前往「Developers」>「API Tokens」。
- 複製現有的正式版 API 權杖，或按一下「產生 API 權杖」建立新的權杖。
- 輸入權杖名稱 (例如
Google Security Operations Integration)。 - 選取「Live」做為權杖類型。
複製並妥善儲存 API 權杖。
驗證權限
如要確認 API 權杖具備必要權限，請按照下列步驟操作：
- 登入 Onfido 資訊主頁。
- 依序前往「開發人員」>「API 權杖」。
- 確認權杖狀態為「Live」，且未撤銷。
- 測試 API 存取權，確認權杖具備檢查和報表的讀取權限。
測試 API 存取權
請先測試憑證，再繼續整合：
# Replace with your actual API token ONFIDO_API_TOKEN="your-api-token" # Test API access - list checks curl -v -H "Authorization: Token token=${ONFIDO_API_TOKEN}" \ "[https://api.onfido.com/v3.6/checks](https://api.onfido.com/v3.6/checks)"
成功的回應會傳回 HTTP 200，並附上包含 checks 陣列的 JSON 物件。
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 onfido-verification-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
建立 Cloud Run 函式的服務帳戶
Cloud Run 函式需要服務帳戶，並具備寫入 GCS bucket 的權限，且可由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
onfido-logs-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Onfido verification logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run 叫用者」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 GCS bucket 的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶：
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
onfido-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。
- 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 提供下列設定詳細資料：
- 主題 ID：輸入
onfido-logs-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Onfido API 擷取驗證檢查和報告，並寫入 GCS。
- 在 GCP 控制台中，前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 onfido-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」或更新版本 在「觸發條件 (選用)」部分：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇
onfido-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁：
- 「服務帳戶」：選取
onfido-logs-collector-sa。
- 「服務帳戶」：選取
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETonfido-verification-logsGCS bucket 名稱 GCS_PREFIXonfido-logs記錄檔的前置字串 STATE_KEYonfido-logs/state.json狀態檔案路徑 ONFIDO_API_TOKENyour-api-token-hereOnfido API 權杖 MAX_RECORDS1000每次執行的記錄數上限 PAGE_SIZE100每頁記錄數 LOOKBACK_HOURS24初始回溯期 在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入
600秒 (10 分鐘)。
- 要求逾時：輸入
前往「設定」分頁：
- 在「Resources」(資源) 部分：
- 記憶體：選取 512 MiB 以上。
- CPU：選取「1」。
- 在「Resources」(資源) 部分：
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入
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', 'onfido-logs') STATE_KEY = os.environ.get('STATE_KEY', 'onfido-logs/state.json') ONFIDO_API_TOKEN = os.environ.get('ONFIDO_API_TOKEN') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) API_BASE = '[https://api.onfido.com/v3.6](https://api.onfido.com/v3.6)' 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) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Onfido verification checks and reports and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ONFIDO_API_TOKEN]): print('Error: Missing required environment variables') return try: # Get GCS bucket 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()}") # Fetch checks checks, newest_check_time = fetch_checks( api_token=ONFIDO_API_TOKEN, start_time=last_time, end_time=now, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) # Fetch reports for each check all_records = [] for check in checks: check_record = check.copy() check_id = check.get('id') if check_id: reports = fetch_reports(api_token=ONFIDO_API_TOKEN, check_id=check_id) check_record['reports'] = reports all_records.append(check_record) if not all_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 all_records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest event time if newest_check_time: save_state(bucket, STATE_KEY, newest_check_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(all_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_checks(api_token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int): """ Fetch verification checks from the Onfido API with pagination and rate limiting. Args: api_token: Onfido API token start_time: Start time for check query end_time: End time for check query page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (checks list, newest_event_time ISO string) """ headers = { 'Authorization': f'Token token={api_token}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-OnfidoCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 current_page = 1 while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break url = f"{API_BASE}/checks?page={current_page}&per_page={page_size}" try: response = http.request('GET', url, 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('checks', []) if not page_results: print(f"No more results (empty page)") break # Filter checks within the time window filtered = [] for check in page_results: created_at = check.get('created_at') if created_at: try: check_time = parse_datetime(created_at) if start_time <= check_time <= end_time: filtered.append(check) if newest_time is None or check_time > parse_datetime(newest_time): newest_time = created_at except Exception as e: print(f"Warning: Could not parse check time: {e}") filtered.append(check) print(f"Page {page_num}: Retrieved {len(page_results)} checks, {len(filtered)} in time window") records.extend(filtered) # Check for more results if len(page_results) < page_size: print(f"Reached last page (size={len(page_results)} < limit={page_size})") break current_page += 1 except Exception as e: print(f"Error fetching checks: {e}") return [], None print(f"Retrieved {len(records)} total checks from {page_num} pages") return records[:max_records], newest_time def fetch_reports(api_token: str, check_id: str): """ Fetch reports for a specific check from the Onfido API. Args: api_token: Onfido API token check_id: Check ID to fetch reports for Returns: List of report objects """ headers = { 'Authorization': f'Token token={api_token}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-OnfidoCollector/1.0' } url = f"{API_BASE}/reports?check_id={check_id}" try: response = http.request('GET', url, headers=headers) if response.status == 429: time.sleep(2) response = http.request('GET', url, headers=headers) if response.status != 200: print(f"Error fetching reports for check {check_id}: HTTP {response.status}") return [] data = json.loads(response.data.decode('utf-8')) reports = data.get('reports', []) return reports except Exception as e: print(f"Error fetching reports for check {check_id}: {e}") return []- 第二個檔案：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」。
點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 onfido-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取「 onfido-logs-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
請根據記錄量和延遲時間需求選擇頻率：
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
高流量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
普通量 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
低量、批次處理 |
| 每天 | 0 0 * * * |
歷來資料集合 |
測試整合
- 在 Cloud Scheduler 控制台中找到工作。
- 按一下「強制執行」，手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「服務」。
- 按一下
onfido-logs-collector。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。尋找：
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X checks, X in time window Wrote X records to gs://bucket-name/onfido-logs/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」 > 「Bucket」。
按一下 bucket 名稱。
前往
onfido-logs/資料夾。確認是否已建立含有目前時間戳記的新
.ndjson檔案。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 Onfido API 權杖
- HTTP 403：確認 API 權杖具有即時存取權，且未撤銷
- HTTP 429：頻率限制 - 函式會自動重試並延遲
- 缺少環境變數：確認已設定所有必要變數
在 Google SecOps 中設定動態饋給，以便擷取 Onfido 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Onfido Verification Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「記錄類型」中選取「Onfido」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址。後續步驟會用到。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket 網址：輸入 GCS bucket URI，並加上前置路徑：
gs://onfido-verification-logs/onfido-logs/- 取代：
onfido-verification-logs：GCS bucket 名稱。onfido-logs：儲存記錄的選用前置字元/資料夾路徑 (留空則為根目錄)。
- 取代：
來源刪除選項：根據偏好選取刪除選項：
- Never：轉移後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
- 刪除已轉移的檔案和空目錄：成功轉移後，刪除檔案和空目錄。
檔案存在時間上限：包含過去幾天內修改過的檔案。預設為 180 天。
資產命名空間：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的新聞動態消息設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶必須具備 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
按一下「授予存取權」。
提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| read_only_udm.metadata.vendor_name | 設為「ONFIDO」。 | |
| read_only_udm.metadata.product_name | 設為「ONFIDO」。 | |
| read_only_udm.metadata.log_type | 設為「ONFIDO」。 |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。