收集 HYPR MFA 記錄
支援的服務：
Google SecOps
SIEM
本文說明如何使用 Webhook 或 Google Cloud Storage V2，將 HYPR MFA 記錄檔擷取至 Google Security Operations。
HYPR MFA 是無密碼多重驗證解決方案，採用 FIDO2 密碼金鑰、生物辨識和行動裝置登入，提供可防範網路釣魚的驗證機制。HYPR 以安全的公開金鑰密碼編譯技術取代傳統密碼，消除以憑證為基礎的攻擊，同時簡化工作站、網頁應用程式和雲端服務的使用者驗證程序。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- HYPR Control Center 的管理員權限
- 與 HYPR 支援團隊聯絡，為要監控的 RP 應用程式啟用自訂事件掛鉤
收集方法差異
HYPR MFA 支援兩種將記錄傳送至 Google Security Operations 的方法：
- Webhook (建議使用)：HYPR 會透過自訂事件掛鉤，將事件即時傳送至 Google Security Operations。這個方法可立即傳送事件，且不需要額外的基礎架構。
- Google Cloud Storage：HYPR 事件會透過 API 收集並儲存在 GCS，然後由 Google Security Operations 擷取。這個方法提供批次處理和歷來資料保留功能。
請選擇最符合需求的方法：
| 功能 | Webhook | Google Cloud Storage |
|---|---|---|
| 延遲 | 即時 (秒) | 批次 (幾分鐘到幾小時) |
| 基礎架構 | 無 | 具備 Cloud Run 函式的 GCP 專案 |
| 歷來資料 | 僅限事件串流 | 在 GCS 中完整保留 |
| 設定複雜度 | 簡單易用 | 中 |
| 費用 | 最低 | GCP 運算和儲存空間費用 |
選項 1：設定 Webhook 整合
在 Google SecOps 中建立 Webhook 動態饋給
建立動態饋給
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態消息」。
- 在下一頁，點按「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
HYPR MFA Events)。 - 「Source type」(來源類型) 請選取「Webhook」。
- 選取「HYPR MFA」做為「記錄類型」。
- 點選「下一步」。
- 指定下列輸入參數的值：
- Split delimiter (選用)：留空。每個 Webhook 要求都包含單一 JSON 事件。
- Asset namespace：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
- 點選「下一步」。
- 在「Finalize」(完成) 畫面中檢查新的新聞動態消息設定，然後按一下「Submit」(提交)。
產生並儲存密鑰
建立動態饋給後，您必須產生用於驗證的密鑰：
- 在動態饋給詳細資料頁面中，按一下「產生密鑰」。
- 對話方塊會顯示密鑰。
- 複製並妥善儲存密鑰。
取得動態饋給端點網址
- 前往動態饋給的「詳細資料」分頁。
- 在「Endpoint Information」部分，複製「Feed endpoint URL」。
網址格式為：
https://{region}-chronicle.googleapis.com/v1/{parent}/logs:import其中
{region}是 Google SecOps 執行個體的區域 (例如us、europe或asia-southeast1)，{parent}是資源路徑，格式為projects/<PROJECT_ID>/locations/<LOCATION>/instances/<INSTANCE_ID>/logTypes/<LOG_TYPE>。請儲存這個網址，以便進行後續步驟。
按一下「完成」。
建立 Google Cloud API 金鑰
Chronicle 需要 API 金鑰才能進行驗證。在 Google Cloud 控制台中建立受限制的 API 金鑰。
建立 API 金鑰
- 前往 Google Cloud 控制台的「憑證」頁面。
- 選取專案 (與 Chronicle 執行個體相關聯的專案)。
- 依序點選「建立憑證」>「API 金鑰」。
- 系統會建立 API 金鑰，並在對話方塊中顯示。
- 按一下「編輯 API 金鑰」，即可限制金鑰。
限制 API 金鑰
- 在「API 金鑰」設定頁面中：
- 名稱：輸入描述性名稱 (例如
Chronicle Webhook API Key)。
- 名稱：輸入描述性名稱 (例如
- 在「API 限制」下方：
- 選取「Restrict key」(限制金鑰)。
- 在「選取 API」下拉式選單中，搜尋並選取「Google SecOps API」 (或「Chronicle API」)。
- 按一下 [儲存]。
- 從頁面頂端的「API 金鑰」欄位複製 API 金鑰值。
- 安全儲存 API 金鑰。
設定 HYPR MFA 自訂事件掛鉤
建構含有標頭的 Webhook 網址
HYPR 支援自訂標頭驗證。使用標頭驗證方式，提升安全性。
端點網址 (不含參數)：
<ENDPOINT_URL>標頭：
x-goog-chronicle-auth: <API_KEY> x-chronicle-auth: <SECRET_KEY>- 取代：
<ENDPOINT_URL>：上一步中的動態饋給端點網址。<API_KEY>：您建立的 Google Cloud API 金鑰。<SECRET_KEY>：建立 Chronicle 動態饋給時產生的 Secret Key。
- 取代：
準備自訂事件掛鉤 JSON 設定
HYPR 自訂事件掛鉤是使用 JSON 設定。準備下列 JSON 設定，並替換預留位置值：
{ "name": "Chronicle SIEM Integration", "eventType": "ALL", "invocationEndpoint": "<ENDPOINT_URL>", "httpMethod": "POST", "authType": "API_KEY", "authParams": { "apiKeyAuthParameters": { "apiKeyName": "x-goog-chronicle-auth", "apiKeyValue": "<API_KEY>" }, "invocationHttpParameters": { "headerParameters": [ { "key": "Content-Type", "value": "application/json", "isValueSecret": false }, { "key": "x-chronicle-auth", "value": "<SECRET_KEY>", "isValueSecret": true } ] } } }更改項目：
<ENDPOINT_URL>：Chronicle 動態饋給端點網址。<API_KEY>：Google Cloud API 金鑰。<SECRET_KEY>：Chronicle 密鑰。
設定參數：
name：事件掛鉤的描述性名稱 (例如
Chronicle SIEM Integration)。eventType：設為
ALL即可傳送所有 HYPR 事件，或指定特定事件標記，例如AUTHENTICATION、REGISTRATION或ACCESS_TOKEN。invocationEndpoint：Chronicle 動態饋給端點網址。
httpMethod：設為
POST。authType：設為
API_KEY，以便使用 API 金鑰驗證。apiKeyName：API 金鑰的標頭名稱 (
x-goog-chronicle-auth)。apiKeyValue：Google Cloud API 金鑰值。
headerParameters：其他標頭，包括
Content-Type: application/json和x-chronicle-auth標頭中的 Chronicle 密鑰。
在 HYPR 控制中心建立自訂事件掛鉤
- 以管理員身分登入 HYPR 控制中心。
- 在左側導覽選單中，按一下 [Integrations] (整合)。
- 在「Integrations」頁面，點選「Add New Integration」。
- HYPR 控制中心會顯示可用的整合功能。
- 在「事件掛鉤」下方，按一下「自訂事件」圖塊。
- 按一下「Add New Event Hook」。
- 在「Add New Event Hook」對話方塊中，將準備好的 JSON 內容貼到文字欄位。
- 按一下「Add Event Hook」。
- HYPR Control Center 會返回「Event Hooks」頁面。
自訂事件掛鉤已設定完成，現在會開始將事件傳送至 Google SecOps。
確認 Webhook 運作正常
查看 HYPR Control Center 事件掛鉤狀態
- 登入 HYPR Control Center。
- 前往「Integrations」。
- 按一下「自訂事件」整合。
- 在「事件掛鉤」表格中，確認事件掛鉤是否已列出。
- 按一下事件掛鉤名稱即可查看詳細資料。
- 確認設定符合您的設定。
查看 Chronicle 資訊提供狀態
- 在 Chronicle 中，依序前往 「SIEM Settings」(SIEM 設定) >「Feeds」(動態饋給)。
- 找出你的 Webhook 動態消息。
- 查看「狀態」欄 (應為「有效」)。
- 查看「Events received」(已接收事件) 計數 (應該會遞增)。
- 檢查「上次成功時間」時間戳記 (應為最近的時間)。
在 Chronicle 中驗證記錄
- 依序前往「搜尋」>「UDM 搜尋」。
請使用下列查詢：
metadata.vendor_name = "HYPR" AND metadata.product_name = "MFA"將時間範圍調整為過去 1 小時。
確認事件顯示在結果中。
驗證方式參考資料
HYPR 自訂事件掛鉤支援多種驗證方式。Chronicle 建議使用自訂標頭進行 API 金鑰驗證。
API 金鑰驗證 (建議用於 Chronicle)
設定：
{ "authType": "API_KEY", "authParams": { "apiKeyAuthParameters": { "apiKeyName": "x-goog-chronicle-auth", "apiKeyValue": "<API_KEY>" }, "invocationHttpParameters": { "headerParameters": [ { "key": "Content-Type", "value": "application/json", "isValueSecret": false }, { "key": "x-chronicle-auth", "value": "<SECRET_KEY>", "isValueSecret": true } ] } } }優點：
- 標頭中傳送的 API 金鑰和密鑰 (比網址參數更安全)。
- 支援多個驗證標頭。
- 標頭未記錄在網路伺服器存取記錄中。
基本驗證
設定：
{ "authType": "BASIC", "authParams": { "basicAuthParameters": { "username": "your-username", "password": "your-password" }, "invocationHttpParameters": { "headerParameters": [ { "key": "Content-Type", "value": "application/json", "isValueSecret": false } ] } } }- 用途：目標系統需要 HTTP 基本驗證時。
OAuth 2.0 用戶端憑證
設定：
{ "authType": "OAUTH_CLIENT_CREDENTIALS", "authParams": { "oauthParameters": { "clientParameters": { "clientId": "your-client-id", "clientSecret": "your-client-secret" }, "authorizationEndpoint": "https://login.example.com/oauth2/v2.0/token", "httpMethod": "POST", "oauthHttpParameters": { "bodyParameters": [ { "key": "scope", "value": "api://your-api/.default", "isValueSecret": false }, { "key": "grant_type", "value": "client_credentials", "isValueSecret": false } ] } }, "invocationHttpParameters": { "headerParameters": [ { "key": "Content-Type", "value": "application/json", "isValueSecret": false } ] } } }- 用途：目標系統需要 OAuth 2.0 驗證。
事件類型和篩選條件
系統會使用 eventTags 參數將 HYPR 事件分組。您可以設定自訂事件掛鉤，傳送所有事件或依特定事件類型篩選。
事件代碼
- 驗證：使用者驗證事件 (登入、解鎖)。
- REGISTRATION：裝置註冊事件 (配對行動裝置、安全金鑰)。
- ACCESS_TOKEN：存取權杖生成和使用事件。
- AUDIT：稽核記錄事件 (管理動作、設定變更)。
設定事件篩選條件
如要只傳送特定事件類型，請修改 JSON 設定中的 eventType 參數：
傳送所有事件：
{ "eventType": "ALL" }僅傳送驗證事件：
{ "eventType": "AUTHENTICATION" }僅傳送註冊事件：
{ "eventType": "REGISTRATION" }
選項 2：設定 Google Cloud Storage 整合功能
整合 GCS 的其他必要條件
除了「事前準備」一節列出的必要條件外，您還需要：
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- HYPR API 憑證 (如要存取 API，請與 HYPR 支援團隊聯絡)
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 hypr-mfa-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或資料保留政策 點選「建立」。
收集 HYPR API 憑證
請與 HYPR 支援團隊聯絡，取得 API 憑證，以便存取 HYPR 事件資料。您需要：
- API 基準網址：HYPR 執行個體網址 (例如
https://your-tenant.hypr.com) - API 權杖：用於存取 API 的驗證權杖
- RP 應用程式 ID：要監控的 Relying Party 應用程式 ID
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要服務帳戶，並具備寫入 GCS bucket 的權限，且可由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 提供下列設定詳細資料：
- 服務帳戶名稱：輸入
hypr-logs-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect HYPR MFA logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions 叫用者」。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 GCS bucket 的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶 (hypr-logs-collector-sa)：
- 依序前往「Cloud Storage」>「Buckets」。
- 點按 bucket 名稱 (例如
hypr-mfa-logs)。 - 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
hypr-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。
- 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 提供下列設定詳細資料：
- 主題 ID：輸入
hypr-logs-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Scheduler 傳送 Pub/Sub 訊息時，會觸發 Cloud Run 函式，從 HYPR API 擷取記錄並寫入 GCS。
- 在 GCP 控制台中，前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 hypr-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件 (選用)」部分：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 Pub/Sub 主題 (
hypr-logs-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (
hypr-logs-collector-sa)。
- 服務帳戶：選取服務帳戶 (
前往「容器」分頁：
- 按一下「Variables & Secrets」(變數與密鑰)。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKEThypr-mfa-logsGCS bucket 名稱 GCS_PREFIXhypr-events記錄檔的前置字串 STATE_KEYhypr-events/state.json狀態檔案路徑 HYPR_API_URLhttps://your-tenant.hypr.comHYPR API 基準網址 HYPR_API_TOKENyour-api-tokenHYPR API 驗證權杖 HYPR_RP_APP_IDyour-rp-app-idHYPR RP 應用程式 ID MAX_RECORDS1000每次執行的記錄數上限 PAGE_SIZE100每頁記錄數 LOOKBACK_HOURS24初始回溯期 在「Variables & Secrets」(變數與密鑰) 部分，向下捲動至「Requests」(要求)：
- 要求逾時：輸入
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 import base64 # 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', 'hypr-events') STATE_KEY = os.environ.get('STATE_KEY', 'hypr-events/state.json') HYPR_API_URL = os.environ.get('HYPR_API_URL') HYPR_API_TOKEN = os.environ.get('HYPR_API_TOKEN') HYPR_RP_APP_ID = os.environ.get('HYPR_RP_APP_ID') 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')) def to_unix_millis(dt: datetime) -> int: """Convert datetime to Unix epoch milliseconds.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return int(dt.timestamp() * 1000) 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 HYPR MFA logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, HYPR_API_URL, HYPR_API_TOKEN, HYPR_RP_APP_ID]): 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()}") # Convert to Unix milliseconds for HYPR API start_millis = to_unix_millis(last_time) end_millis = to_unix_millis(now) # Fetch logs records, newest_event_time = fetch_logs( api_url=HYPR_API_URL, api_token=HYPR_API_TOKEN, rp_app_id=HYPR_RP_APP_ID, start_time_ms=start_millis, end_time_ms=end_millis, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) if not 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 records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest event time if newest_event_time: save_state(bucket, STATE_KEY, newest_event_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(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_logs(api_url: str, api_token: str, rp_app_id: str, start_time_ms: int, end_time_ms: int, page_size: int, max_records: int): """ Fetch logs from HYPR API with pagination and rate limiting. Args: api_url: HYPR API base URL api_token: HYPR API authentication token rp_app_id: HYPR RP application ID start_time_ms: Start time in Unix milliseconds end_time_ms: End time in Unix milliseconds page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ # Clean up API URL base_url = api_url.rstrip('/') endpoint = f"{base_url}/rp/api/versioned/events" # Bearer token authentication headers = { 'Authorization': f'Bearer {api_token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-HYPRCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 # Offset-based pagination start_index = 0 while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break # Build request parameters params = [] params.append(f"rpAppId={rp_app_id}") params.append(f"startDate={start_time_ms}") params.append(f"endDate={end_time_ms}") params.append(f"start={start_index}") params.append(f"limit={min(page_size, max_records - len(records))}") url = f"{endpoint}?{'&'.join(params)}" 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')) # Extract results page_results = data.get('data', []) if not page_results: print(f"No more results (empty page)") break print(f"Page {page_num}: Retrieved {len(page_results)} events") records.extend(page_results) # Track newest event time for event in page_results: try: # HYPR uses LOGGEDTIMEINUTC field with Unix milliseconds event_time_ms = event.get('LOGGEDTIMEINUTC') if event_time_ms: event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc) event_time = event_dt.isoformat() if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for more results current_size = data.get('size', 0) if current_size < page_size: print(f"Reached last page (size={current_size} < limit={page_size})") break start_index += current_size except Exception as e: print(f"Error fetching logs: {e}") return [], None print(f"Retrieved {len(records)} total records from {page_num} pages") return records, newest_time- 第二個檔案：requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0按一下「部署」，即可儲存並部署函式。
等待部署作業完成 (2 至 3 分鐘)。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題 (hypr-logs-trigger)，觸發 Cloud Run 函式。
- 前往 GCP 控制台中的「Cloud Scheduler」。
- 點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 hypr-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議選取世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 ( hypr-logs-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
請根據記錄量和延遲時間需求選擇頻率：
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
高流量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等音量 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
低量、批次處理 |
| 每天 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中，找到您的工作 (
hypr-logs-collector-hourly)。 - 按一下「強制執行」，手動觸發工作。
- 稍等幾秒鐘。
- 依序前往「Cloud Run」 > 「服務」。
- 按一下函式名稱 (
hypr-logs-collector)。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。尋找：
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X events Wrote X records to gs://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」 > 「Bucket」。
點選 bucket 名稱 (例如
hypr-mfa-logs)。前往前置字元資料夾 (例如
hypr-events/)。確認是否已建立含有目前時間戳記的新
.ndjson檔案。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 API 憑證
- HTTP 403：確認 HYPR API 權杖具備必要權限，且 RP 應用程式 ID 正確無誤
- HTTP 429：頻率限制 - 函式會自動重試並延遲
- 缺少環境變數：確認已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
在 Google SecOps 中設定動態饋給，擷取 HYPR MFA 記錄
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
HYPR MFA Logs from GCS)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
選取「HYPR MFA」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket 網址：輸入 GCS bucket URI，並加上前置路徑：
gs://hypr-mfa-logs/hypr-events/- 取代：
hypr-mfa-logs：GCS bucket 名稱。hypr-events：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。
- 取代：
來源刪除選項：根據偏好選取刪除選項：
- 永不：一律不會在轉移後刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空目錄：成功轉移後，刪除檔案和空目錄。
檔案存在時間上限：納入在過去天數內修改的檔案。預設為 180 天。
資產命名空間：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的新聞動態消息設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶必須具備 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱 (例如
hypr-mfa-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
- 按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| extensions.auth.type | 驗證類型 (例如單一登入、多重驗證) | |
| metadata.event_type | 事件類型 (例如 USER_LOGIN、NETWORK_CONNECTION) | |
| EVENTNAME | metadata.product_event_type | 產品專屬事件類型 |
| ID | metadata.product_log_id | 產品專屬記錄 ID |
| USERAGENT | network.http.parsed_user_agent | 已剖析的 HTTP 使用者代理程式 |
| USERAGENT | network.http.user_agent | HTTP 使用者代理程式字串 |
| SESSIONID | network.session_id | 工作階段 ID |
| DEVICEMODEL | principal.asset.hardware.model | 資產的硬體型號 |
| COMPANION,MACHINEDOMAIN | principal.asset.hostname | 資產的主機名稱 |
| REMOTEIP | principal.asset.ip | 資產的 IP 位址 |
| DEVICEID | principal.asset_id | 資產的專屬 ID |
| COMPANION,MACHINEDOMAIN | principal.hostname | 與主體相關聯的主機名稱 |
| REMOTEIP | principal.ip | 與主體相關聯的 IP 位址 |
| DEVICEOS | principal.platform | 平台 (例如 WINDOWS、LINUX) |
| DEVICEOSVERSION | principal.platform_version | 平台版本 |
| ISSUCCESSFUL | security_result.action | 保全系統採取的動作 (例如 ALLOW、BLOCK) |
| 訊息 | security_result.description | 安全性結果說明 |
| MACHINEUSERNAME | target.user.user_display_name | 使用者的顯示名稱 |
| FIDOUSER | target.user.userid | 使用者 ID |
| metadata.product_name | 產品名稱 | |
| metadata.vendor_name | 供應商/公司名稱 |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專家尋求解答。