收集 BeyondTrust Endpoint Privilege Management (EPM) 記錄
本文說明如何使用 Google Cloud Storage,將 BeyondTrust Endpoint Privilege Management (EPM) 記錄檔擷取至 Google Security Operations。剖析器主要負責將 BeyondTrust Endpoint 的原始 JSON 記錄資料轉換為符合 Chronicle UDM 的結構化格式。這個函式會先初始化各種欄位的預設值,然後剖析 JSON 酬載,接著將原始記錄中的特定欄位對應至 event.idm.read_only_udm 物件中的對應 UDM 欄位。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- BeyondTrust Endpoint Privilege Management 租戶或 API 的特殊存取權
建立 Google Cloud Storage 值區
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 beyondtrust-epm-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
收集 BeyondTrust EPM API 憑證
- 以管理員身分登入 BeyondTrust Privilege Management 網頁控制台。
- 依序前往「設定」>「設定」>「API 設定」。
- 按一下「Create an API Account」(建立 API 帳戶)。
- 請提供下列設定詳細資料:
- 「Name」(名稱):輸入
Google SecOps Collector。 - API 存取權:啟用「稽核 (讀取)」和其他必要範圍。
- 「Name」(名稱):輸入
- 複製並儲存「用戶端 ID」和「用戶端密鑰」。
- 複製 API 基礎網址,通常是
https://<your-tenant>-services.pm.beyondtrustcloud.com(您會將此網址做為 BPT_API_URL 使用)。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶,並由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 主控台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下 [Create Service Account] (建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
beyondtrust-epm-collector-sa。 - 服務帳戶說明:輸入
Service account for Cloud Run function to collect BeyondTrust EPM 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 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
beyondtrust-epm-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 BeyondTrust EPM API 擷取記錄,並將記錄寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 beyondtrust-epm-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇主題
beyondtrust-epm-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶
beyondtrust-epm-collector-sa。
- 服務帳戶:選取服務帳戶
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 GCS_BUCKETbeyondtrust-epm-logsGCS_PREFIXbeyondtrust-epm/STATE_KEYbeyondtrust-epm/state.jsonBPT_API_URLhttps://yourtenant-services.pm.beyondtrustcloud.comCLIENT_IDyour-client-idCLIENT_SECRETyour-client-secretOAUTH_SCOPEmanagement-apiRECORD_SIZE1000MAX_ITERATIONS10在「變數與密鑰」分頁中向下捲動至「要求」:
- 要求逾時:輸入
600秒 (10 分鐘)。
- 要求逾時:輸入
前往「容器」中的「設定」分頁:
- 在「資源」部分:
- 記憶體:選取 512 MiB 以上。
- CPU:選取 1。
- 按一下 [完成]。
- 在「資源」部分:
捲動至「執行環境」:
- 選取「預設」 (建議選項)。
在「修訂版本資源調度」部分:
- 執行個體數量下限:輸入
0。 - 「Maximum number of instances」(執行個體數量上限):輸入
100(或根據預期負載調整)。
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「Function entry point」(函式進入點) 中輸入 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 http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from BeyondTrust EPM API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/') state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm/state.json') # BeyondTrust EPM API credentials api_url = os.environ.get('BPT_API_URL') client_id = os.environ.get('CLIENT_ID') client_secret = os.environ.get('CLIENT_SECRET') oauth_scope = os.environ.get('OAUTH_SCOPE', 'management-api') record_size = int(os.environ.get('RECORD_SIZE', '1000')) max_iterations = int(os.environ.get('MAX_ITERATIONS', '10')) if not all([bucket_name, api_url, client_id, client_secret]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Load state (last processed timestamp) state = load_state(bucket, state_key) last_timestamp = state.get('last_timestamp', (datetime.utcnow() - timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ")) print(f'Processing logs since {last_timestamp}') # Get OAuth access token token = get_oauth_token(api_url, client_id, client_secret, oauth_scope) # Fetch audit events events = fetch_audit_events(api_url, token, last_timestamp, record_size, max_iterations) if events: # Store events in GCS current_timestamp = datetime.utcnow() filename = f"{prefix}beyondtrust-epm-events-{current_timestamp.strftime('%Y%m%d_%H%M%S')}.json" store_events_to_gcs(bucket, filename, events) # Update state with latest timestamp latest_timestamp = get_latest_event_timestamp(events) save_state(bucket, state_key, {'last_timestamp': latest_timestamp, 'updated_at': datetime.utcnow().isoformat() + 'Z'}) print(f'Successfully processed {len(events)} events and stored to {filename}') else: print('No new events found') except Exception as e: print(f'Error processing logs: {str(e)}') raise def get_oauth_token(api_url, client_id, client_secret, scope): """ Get OAuth access token using client credentials flow for BeyondTrust EPM. Uses the correct endpoint: /oauth/token """ token_url = f"{api_url}/oauth/token" headers = {'Content-Type': 'application/x-www-form-urlencoded'} body = f"grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}&scope={scope}" response = http.request('POST', token_url, headers=headers, body=body, timeout=urllib3.Timeout(60.0)) if response.status != 200: raise RuntimeError(f"Token request failed: {response.status} {response.data[:256]!r}") token_data = json.loads(response.data.decode('utf-8')) return token_data['access_token'] def fetch_audit_events(api_url, access_token, last_timestamp, record_size, max_iterations): """ Fetch audit events using the BeyondTrust EPM API endpoint: /management-api/v2/AuditEvents with query parameters for filtering and pagination """ headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } all_events = [] current_start_date = last_timestamp iterations = 0 # Enforce maximum RecordSize limit of 1000 based on BeyondTrust documentation record_size_limited = min(record_size, 1000) while iterations < max_iterations: iterations += 1 if len(all_events) >= 10000: print(f"Reached maximum events limit (10000)") break # Use the BeyondTrust EPM API endpoint for audit events query_url = f"{api_url}/management-api/v2/AuditEvents" params = { 'StartDate': current_start_date, 'RecordSize': record_size_limited } # Construct URL with query parameters query_string = '&'.join([f"{k}={v}" for k, v in params.items()]) full_url = f"{query_url}?{query_string}" try: response = http.request('GET', full_url, headers=headers, timeout=urllib3.Timeout(300.0)) # Handle rate limiting with exponential backoff if response.status == 429: retry_after = int(response.headers.get('Retry-After', '30')) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) continue if response.status != 200: raise RuntimeError(f"API request failed: {response.status} {response.data[:256]!r}") response_data = json.loads(response.data.decode('utf-8')) events = response_data.get('events', []) if not events: break print(f"Page {iterations}: Retrieved {len(events)} events") all_events.extend(events) # If we got fewer events than RecordSize, we've reached the end if len(events) < record_size_limited: break # For pagination, update StartDate to the timestamp of the last event last_event = events[-1] last_timestamp = extract_event_timestamp(last_event) if not last_timestamp: print("Warning: Could not find timestamp in last event for pagination") break # Convert to datetime and add 1 second to avoid retrieving the same event again try: dt = parse_timestamp(last_timestamp) dt = dt + timedelta(seconds=1) current_start_date = dt.strftime("%Y-%m-%dT%H:%M:%SZ") except Exception as e: print(f"Error parsing timestamp {last_timestamp}: {e}") break except Exception as e: print(f"Error fetching page {iterations}: {e}") break return all_events def extract_event_timestamp(event): """Extract timestamp from event, checking multiple possible fields""" # Check common timestamp fields timestamp_fields = ['event.dateTime', 'event.timestamp', 'timestamp', 'eventTime', 'dateTime', 'whenOccurred', 'date', 'time', 'event.ingested'] # Try nested event.dateTime first (common in BeyondTrust) if isinstance(event, dict) and isinstance(event.get("event"), dict): ts = event["event"].get("dateTime") if ts: return ts ts = event["event"].get("timestamp") if ts: return ts # Fallback to other timestamp fields for field in timestamp_fields: if field in event and event[field]: return event[field] return None def parse_timestamp(timestamp_str): """Parse timestamp string to datetime object, handling various formats""" if isinstance(timestamp_str, (int, float)): # Unix timestamp (in milliseconds or seconds) if timestamp_str > 1e12: # Milliseconds return datetime.fromtimestamp(timestamp_str / 1000, tz=timezone.utc) else: # Seconds return datetime.fromtimestamp(timestamp_str, tz=timezone.utc) if isinstance(timestamp_str, str): # Try different string formats try: # ISO format with Z if timestamp_str.endswith('Z'): return datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) # ISO format with timezone elif '+' in timestamp_str or timestamp_str.endswith('00:00'): return datetime.fromisoformat(timestamp_str) # ISO format without timezone (assume UTC) else: dt = datetime.fromisoformat(timestamp_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt except ValueError: pass raise ValueError(f"Could not parse timestamp: {timestamp_str}") 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: {str(e)}') return {} def save_state(bucket, key, state): """Save state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}') def store_events_to_gcs(bucket, key, events): """Store events as JSONL (one JSON object per line) in GCS""" # Convert to JSONL format (one JSON object per line) jsonl_content = '\n'.join(json.dumps(event, default=str) for event in events) blob = bucket.blob(key) blob.upload_from_string(jsonl_content, content_type='application/x-ndjson') def get_latest_event_timestamp(events): """Get the latest timestamp from the events for state tracking""" if not events: return datetime.utcnow().isoformat() + 'Z' latest = None for event in events: timestamp = extract_event_timestamp(event) if timestamp: try: event_dt = parse_timestamp(timestamp) event_iso = event_dt.isoformat() + 'Z' if latest is None or event_iso > latest: latest = event_iso except Exception as e: print(f"Error parsing event timestamp {timestamp}: {e}") continue return latest or datetime.utcnow().isoformat() + 'Z'- 第二個檔案: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」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 beyondtrust-epm-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 beyondtrust-epm-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *高容量、低延遲 每 15 分鐘檢查一次 */15 * * * *普通量 每小時 0 * * * *標準 (建議採用) 每 6 小時 0 */6 * * *少量、批次處理 每日 0 0 * * *歷來資料集合
測試排程器工作
- 在 Cloud Scheduler 控制台中找出您的工作。
- 按一下「強制執行」即可手動觸發。
- 等待幾秒鐘,然後依序前往「Cloud Run」>「Services」>「beyondtrust-epm-collector」>「Logs」。
- 確認函式是否已順利執行。
- 檢查 GCS 值區,確認是否已寫入記錄。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
BeyondTrust EPM logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「BeyondTrust Endpoint Privilege Management」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示不重複的服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定動態饋給,擷取 BeyondTrust EPM 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
BeyondTrust EPM logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「BeyondTrust Endpoint Privilege Management」做為「記錄類型」。
- 點選 [下一步]。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://beyondtrust-epm-logs/beyondtrust-epm/取代:
beyondtrust-epm-logs:您的 GCS bucket 名稱。beyondtrust-epm/:儲存記錄的選用前置字元/資料夾路徑 (如為根目錄,請留空)。
範例:
- 根層級 bucket:
gs://beyondtrust-epm-logs/ - 前置字串:
gs://beyondtrust-epm-logs/beyondtrust-epm/
- 根層級 bucket:
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| agent.id | principal.asset.attribute.labels.value | 對應至含有鍵 agent_id 的標籤 |
| agent.version | principal.asset.attribute.labels.value | 對應至含有鍵 agent_version 的標籤 |
| ecs.version | principal.asset.attribute.labels.value | 對應至具有金鑰 ecs_version 的標籤 |
| event_data.reason | metadata.description | 原始記錄中的事件說明 |
| event_datas.ActionId | metadata.product_log_id | 產品專屬記錄 ID |
| file.path | principal.file.full_path | 活動中的完整檔案路徑 |
| headers.content_length | additional.fields.value.string_value | 已對應至具有重要 content_length 的標籤 |
| headers.content_type | additional.fields.value.string_value | 對應至含有主要 content_type 的標籤 |
| headers.http_host | additional.fields.value.string_value | 對應至含有鍵 http_host 的標籤 |
| headers.http_version | network.application_protocol_version | HTTP 通訊協定版本 |
| headers.request_method | network.http.method | HTTP 要求方法 |
| host.hostname | principal.hostname | 主要主機名稱 |
| host.hostname | principal.asset.hostname | 主要資產主機名稱 |
| host.ip | principal.asset.ip | 主要資產 IP 位址 |
| host.ip | principal.ip | 主要 IP 位址 |
| host.mac | principal.mac | 主要 MAC 位址 |
| host.os.platform | principal.platform | 如果等於 macOS,請設為 MAC |
| host.os.version | principal.platform_version | 作業系統版本 |
| labels.related_item_id | metadata.product_log_id | 相關項目 ID |
| process.command_line | principal.process.command_line | 程序指令列 |
| process.name | additional.fields.value.string_value | 對應至含有鍵 process_name 的標籤 |
| process.parent.name | additional.fields.value.string_value | 對應至含有鍵 process_parent_name 的標籤 |
| process.parent.pid | principal.process.parent_process.pid | 父項程序 PID 轉換為字串 |
| process.pid | principal.process.pid | 程序 PID 轉換為字串 |
| user.id | principal.user.userid | 使用者 ID |
| user.name | principal.user.user_display_name | 使用者顯示名稱 |
| 不適用 | metadata.event_timestamp | 事件時間戳記設為記錄項目時間戳記 |
| 不適用 | metadata.event_type | 如果沒有主體,則為 GENERIC_EVENT,否則為 STATUS_UPDATE |
| 不適用 | network.application_protocol | 如果 http_version 欄位包含 HTTP,請設為 HTTP |
需要其他協助嗎?向社群成員和 Google SecOps 專業人員尋求答案。