收集 CyberArk Endpoint Privilege Manager 記錄
剖析器版本：2.0
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Cloud Storage V2，將 CyberArk Endpoint Privilege Manager (EPM) 記錄檔擷取至 Google Security Operations。
CyberArk Endpoint Privilege Manager 會移除本機管理員權限、控管可執行或提升權限的應用程式，並防止憑證遭竊，在 Windows、macOS 和 Linux 端點上強制執行最小權限原則。EPM SaaS 會透過 EPM REST API 公開事件資料。Cloud Run 函式會輪詢每個 EPM 集的原始事件，並將這些事件寫入 Cloud Storage bucket，Google SecOps 則會從該處擷取事件。
事前準備
請確認您符合下列必要條件：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 Google Cloud 專案
- 建立及管理 Cloud Storage bucket 的權限
- 管理 Cloud Storage 值區身分與存取權管理 (IAM) 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作所需的權限
- CyberArk EPM 軟體即服務 (SaaS) 租戶
- EPM 帳戶管理員，負責建立 API 使用者和角色
建立 Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 cyberark-epm-secops-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
收集 CyberArk EPM 憑證
Cloud Run 函式會以 EPM 使用者身分登入 EPM REST API。建立專用使用者，並授予僅供檢視存取權，讓其存取要收集事件的組合。
建立 API 使用者
- 以帳戶管理員身分登入 EPM 管理控制台。
- 按一下「管理」圖塊。
- 前往「使用者」頁面，然後按一下「新增使用者」。
- 請提供下列設定詳細資料：
- 電子郵件地址：輸入整合使用者的登入名稱 (例如
secops-collector@example.com)。 - 密碼：輸入 12 到 128 個字元的密碼。
- 電子郵件地址：輸入整合使用者的登入名稱 (例如
- 按一下「新增」。
- 安全儲存登入名稱和密碼。您會將這些值做為 Cloud Run 函式的
EPM_USERNAME和EPM_PASSWORD值。
建立並指派「僅供檢視」角色
- 在「管理」中，前往「角色」頁面，然後按一下「建立角色」。
- 請提供下列設定詳細資料：
- 名稱：輸入角色名稱 (例如
SecOps view only)。 - 權限群組：選取「僅供檢視」。
- 組合：選取要收集事件的所有組合。
- 名稱：輸入角色名稱 (例如
- 點選「建立」。
- 前往「角色指派」頁面。
- 在您建立的角色所在資料列中，依序點選「更多動作 (...)」>「指派或取消指派使用者」。
- 按一下「指派使用者」，搜尋並選取整合使用者，然後按一下「指派」。
按一下 [儲存]。
判斷 EPM 登入網址
EPM 驗證 API 是由您所在區域的調度器伺服器提供。函式只會使用這個網址登入，其他呼叫都會傳送至登入回應傳回的管理員網址。
| 區域 | 登入網址 |
|---|---|
| 美國 | https://login.epm.cyberark.com |
| 德國 | https://eu.epm.cyberark.com |
| 瑞士 | https://ch.epm.cyberark.com |
| 英國 | https://uk.epm.cyberark.com |
| 澳洲 | https://au.epm.cyberark.com |
| 加拿大 | https://ca.epm.cyberark.com |
| 印度 | https://in.epm.cyberark.com |
| 日本 | https://jp.epm.cyberark.com |
| 新加坡 | https://sg.epm.cyberark.com |
| 義大利 | https://it.epm.cyberark.com |
| 阿拉伯聯合大公國 | https://ae.epm.cyberark.com |
| 聯邦 | https://login.epm.cyberarkgov.cloud |
測試 API 存取權
請先測試憑證，再繼續進行整合，步驟如下：
# Replace with your actual values EPM_LOGIN_URL="https://login.epm.cyberark.com" EPM_USERNAME="<your-username>" EPM_PASSWORD="<your-password>" # Sign in and get a session token and the manager URL curl -s -X POST "$EPM_LOGIN_URL/EPM/API/Auth/EPM/Logon" \ -H "Content-Type: application/json" \ -d "{\"Username\":\"$EPM_USERNAME\",\"Password\":\"$EPM_PASSWORD\",\"ApplicationID\":\"GoogleSecOps\"}"成功的回應會傳回
EPMAuthenticationResult(工作階段符記) 和ManagerURL。如要確認角色，請使用權杖列出集合，如下所示：# Replace with the values from the sign-in response MANAGER_URL="<ManagerURL>" TOKEN="<EPMAuthenticationResult>" curl -s "$MANAGER_URL/EPM/API/Sets?Limit=1000" \ -H "Authorization: basic $TOKEN"
回應會列出使用者可查看的所有集合。如果 Sets 為空白，表示未指派角色或未包含集合。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備權限的服務帳戶，才能寫入 Cloud Storage bucket 並由 Pub/Sub 叫用。
建立服務帳戶
- 在 Google Cloud 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 提供下列設定詳細資料：
- 服務帳戶名稱：輸入
cyberark-epm-collector-sa - 服務帳戶說明：輸入
Service account for Cloud Run function to collect CyberArk EPM logs
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run 叫用者」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 Cloud Storage bucket，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 Cloud Storage bucket 的 IAM 權限
將 Cloud Storage bucket 的寫入權限授予服務帳戶：
- 在Google Cloud 控制台中，依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
cyberark-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色：選取「Storage 物件管理員」
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。
- 在 Google Cloud 控制台中，依序前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 請提供下列設定詳細資料：
- 主題 ID：輸入
cyberark-epm-trigger - 其他設定保留預設值
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 CyberArk EPM REST API 擷取原始事件，並將這些事件寫入 Cloud Storage。
- 前往Google Cloud 控制台的「Cloud Run」。
- 點選「編寫函式」。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 cyberark-epm-collector區域 選取與 Cloud Storage bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件」部分中，按一下「新增觸發條件」，然後選取「Pub/Sub 觸發條件」。
在「Eventarc trigger」(Eventarc 觸發條件) 窗格中，提供下列設定詳細資料：
- 觸發條件名稱：保留系統產生的名稱，或輸入觸發條件的名稱。
- 觸發條件類型：選取「Google 來源」。
- 事件供應商：選取「Pub/Sub」。
- 事件類型：選取「google.cloud.pubsub.topic.v1.messagePublished」。
- 「選取 Cloud Pub/Sub 主題」：選擇主題
cyberark-epm-trigger。 - 區域：選取與函式相同的區域。
- 服務帳戶：選取服務帳戶
cyberark-epm-collector-sa。
按一下「儲存觸發條件」。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 選取「Identity and Access Management (IAM)」。
前往並展開「容器、網路、安全性」。
前往「安全性」分頁標籤：
- 服務帳戶：選取服務帳戶
cyberark-epm-collector-sa
- 服務帳戶：選取服務帳戶
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETcyberark-epm-secops-logsCloud Storage Bucket 名稱 GCS_PREFIXcyberark-epm記錄檔的前置字串 STATE_KEYcyberark-epm-state.json狀態路徑 (記錄前置字元以外) EPM_LOGIN_URLhttps://login.epm.cyberark.com您所在區域的 EPM 登入網址 EPM_USERNAMEsecops-collector@example.comEPM API 使用者的登入名稱 EPM_PASSWORD<your-password>EPM API 使用者的密碼 EPM_APPLICATION_IDGoogleSecOps登入時回報的應用程式名稱 EPM_SET_IDS(空白) 以半形逗號分隔的選用 ID 集；空白則會收集使用者可查看的所有 ID 集 EPM_API_VERSION(空白) 選用 API 版本 (例如 24.12.0.1)；留空則使用最新版本MAX_RECORDS10000每次執行的記錄數上限 PAGE_SIZE1000每頁記錄數 (API 最多為 1000) LOOKBACK_HOURS24首次執行時的初始回溯期 OVERLAP_MINUTES15在浮水印前重新讀取時間範圍，以便掌握代理商延遲回報的事件 在「變數與密鑰」部分，前往「要求」：
- 要求逾時：輸入
600秒 (10 分鐘)
- 要求逾時：輸入
前往「設定」分頁：
- 在「資源」部分：
- 記憶體：選取 512 MiB 以上
- 「CPU」：選取「1」
- 在「資源」部分：
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入
0 - 執行個體數量上限：輸入
1
- 執行個體數量下限：輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立後，內嵌程式碼編輯器會自動開啟。
新增函式程式碼
- 在「進入點」欄位中輸入 main。
在內嵌程式碼編輯器中，建立兩個檔案：
第一個檔案 - main.py:
import functions_framework from google.cloud import storage from google.cloud.exceptions import NotFound import hashlib import json import os import urllib3 from urllib.parse import urlencode from datetime import datetime, timezone, timedelta import time # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=60.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', 'cyberark-epm') # STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under # its bucket URI and, with a deletion option selected, deletes what it # transferred. A state file inside the prefix would be ingested as log data and # then deleted, resetting collection and re-ingesting duplicates. STATE_KEY = os.environ.get('STATE_KEY', 'cyberark-epm-state.json') EPM_LOGIN_URL = os.environ.get('EPM_LOGIN_URL', 'https://login.epm.cyberark.com') EPM_USERNAME = os.environ.get('EPM_USERNAME') EPM_PASSWORD = os.environ.get('EPM_PASSWORD') EPM_APPLICATION_ID = os.environ.get('EPM_APPLICATION_ID', 'GoogleSecOps') # Comma-separated set IDs to collect. Empty collects every set the user can view. EPM_SET_IDS = [s.strip() for s in os.environ.get('EPM_SET_IDS', '').split(',') if s.strip()] # API version segment. Empty uses the latest version, which CyberArk recommends. EPM_API_VERSION = os.environ.get('EPM_API_VERSION', '').strip() MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000')) # The Events/Search API returns at most 1,000 records per page. PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '1000')), 1000) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) # The query re-reads this many minutes before the watermark. EPM agents report # in batches, so an event can reach the service well after its event date. # Re-read events are removed by content hash. OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '15')) # The events APIs allow 5 calls per second; keep a small gap between calls. CALL_INTERVAL_SECONDS = 0.25 class FetchError(Exception): """Raised when the EPM API call fails. The watermark must never advance on a failed fetch, otherwise the events in the failed window are skipped permanently. """ def parse_datetime(value: str) -> datetime: """Parse an ISO 8601 datetime string into an aware datetime.""" if value.endswith('Z'): value = value[:-1] + '+00:00' dt = datetime.fromisoformat(value) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def epm_time(dt: datetime) -> str: """Render a datetime in the YYYY-MM-DDThh:mm:ssZ form the EPM filter accepts.""" return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') def api_path(*parts: str) -> str: """Build an /EPM/API path, inserting the version segment when one is set.""" segments = ['EPM', 'API'] if EPM_API_VERSION: segments.append(EPM_API_VERSION) segments.extend(parts) return '/' + '/'.join(segments) def event_keys(event: dict) -> set: """Return the identity this event can be recognised by. Raw events carry no unique event ID, so a record counts as already collected when its full JSON content was seen before. The overlap window returns the same JSON again, which this catches without any knowledge of the payload schema. """ return {'sha256:' + hashlib.sha256( json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8') ).hexdigest()} def event_time(event: dict) -> str: """Return the event timestamp as an ISO 8601 string, or an empty string. Raw events carry the time of the most recent occurrence in `lastEventDate`. """ value = event.get('lastEventDate') return str(value) if value else '' @functions_framework.cloud_event def main(cloud_event): """Fetch CyberArk EPM raw events and write them to Cloud Storage. Args: cloud_event: CloudEvent object containing the Pub/Sub message. """ if not all([GCS_BUCKET, EPM_LOGIN_URL, EPM_USERNAME, EPM_PASSWORD]): # Raise rather than return: a bare return acks the Pub/Sub message and # reports the run as successful, silently discarding the schedule tick. raise RuntimeError('Missing required environment variables') bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket, STATE_KEY) set_states = state.get('sets', {}) now = datetime.now(timezone.utc) manager_url, token = login() headers = { 'Authorization': f'basic {token}', 'Content-Type': 'application/json', 'Accept': 'application/json', } sets = list_sets(manager_url, headers) if EPM_SET_IDS: sets = [s for s in sets if s['Id'] in EPM_SET_IDS] missing = set(EPM_SET_IDS) - {s['Id'] for s in sets} if missing: raise FetchError(f'Configured set IDs not visible to this user: {sorted(missing)}') if not sets: raise FetchError('No EPM sets are visible to this user; check the assigned role') fresh_all = [] new_set_states = {} budget = MAX_RECORDS for epm_set in sets: set_id = epm_set['Id'] set_state = set_states.get(set_id, {}) watermark = None if set_state.get('last_event_time'): watermark = parse_datetime(set_state['last_event_time']) seen_keys = set(set_state.get('seen_keys', [])) if watermark is None: start_time = now - timedelta(hours=LOOKBACK_HOURS) else: start_time = watermark - timedelta(minutes=OVERLAP_MINUTES) print(f"Set {set_id} ({epm_set.get('Name', '')}): fetching events from {start_time.isoformat()} to {now.isoformat()}") # A FetchError here propagates: the run fails, the watermark is untouched, # and the next invocation retries the same window. records, newest_event_time = fetch_events( manager_url=manager_url, headers=headers, set_id=set_id, start_time=start_time, end_time=now, page_size=PAGE_SIZE, max_records=budget, ) budget -= len(records) # Remove events already written by an earlier run. Without this, the overlap # window re-emits its events on every invocation. fresh = [r for r in records if not (event_keys(r) & seen_keys)] print(f"Set {set_id}: fetched {len(records)} records, {len(fresh)} new after deduplication") if not fresh: new_set_states[set_id] = set_state continue if not newest_event_time: # A watermark that cannot be derived from the data is not a watermark. # Advancing on wall-clock here would skip every late-reported event. raise FetchError(f'Set {set_id}: records were returned but no event timestamp could be parsed') # The parser reads the set from the event itself. Annotate copies so the # deduplication hashes keep matching what the API returns next time. fresh_all.extend({**record, 'setID': set_id} for record in fresh) new_watermark = parse_datetime(newest_event_time) if watermark and new_watermark < watermark: new_watermark = watermark # Retain only the hashes still inside the overlap window, so the state file # stays small while covering every event the next query can re-read. cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES) retained = set() for record in records: stamp = event_time(record) if not stamp: continue try: if parse_datetime(stamp) >= cutoff: retained.update(event_keys(record)) except ValueError: continue new_set_states[set_id] = { 'last_event_time': new_watermark.isoformat(), 'seen_keys': sorted(retained), } if budget <= 0: print(f"Reached max_records limit ({MAX_RECORDS}); remaining sets are collected by the next run") break # Keep the state of sets not visited in this run. for set_id, set_state in set_states.items(): new_set_states.setdefault(set_id, set_state) if not fresh_all: print("No new events found. Watermarks left unchanged.") return timestamp = now.strftime('%Y%m%dT%H%M%SZ') object_key = f"{GCS_PREFIX}/events_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh_all) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(fresh_all)} records to gs://{GCS_BUCKET}/{object_key}") # Advance the watermarks only after the data is durably written. save_state(bucket, STATE_KEY, {'sets': new_set_states}) print(f"Successfully processed {len(fresh_all)} records") def load_state(bucket, key): """Read the collector state from Cloud Storage. Only a missing object is treated as a cold start. Any other error is raised: swallowing it would silently reset collection to the full lookback window and re-ingest that entire period. """ blob = bucket.blob(key) try: return json.loads(blob.download_as_text()) except NotFound: print('No state file found. Starting from the lookback window.') return {} def save_state(bucket, key, state: dict): """Write the collector state to Cloud Storage. Failures are raised, not logged. If the state write fails after the data was uploaded, the next run repeats the same window and duplicates it. """ blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json', ) print(f"Saved state for {len(state.get('sets', {}))} sets") def request_json(method: str, url: str, headers: dict, body: dict = None): """Call the EPM API and return the decoded JSON body. Raises FetchError on any transport or HTTP failure, so that the caller does not mistake a failed call for an empty result and advance the watermark. """ backoff = 1.0 for attempt in range(5): time.sleep(CALL_INTERVAL_SECONDS) try: response = http.request( method, url, body=json.dumps(body).encode('utf-8') if body is not None else None, headers=headers, ) except Exception as e: raise FetchError(f'Request to {url} failed: {e}') from e if response.status == 429: raw_retry_after = response.headers.get('Retry-After') try: # Retry-After may also be an HTTP date, which int() cannot parse. wait = int(raw_retry_after) if raw_retry_after else int(backoff) except (TypeError, ValueError): wait = int(backoff) print(f"Rate limited (429). Retrying after {wait}s...") time.sleep(wait) backoff = min(backoff * 2, 30.0) continue if response.status != 200: raise FetchError(f'HTTP {response.status} from {url}: {response.data.decode("utf-8", "replace")}') try: return json.loads(response.data.decode('utf-8')) except json.JSONDecodeError as e: raise FetchError(f'Malformed JSON response from {url}: {e}') from e raise FetchError('Rate limited repeatedly; giving up without advancing the watermark') def login(): """Authenticate to EPM and return the manager URL and session token. EPM allows one API login per minute per user, so the function signs in once per run and reuses the token for every call in that run. """ url = f"{EPM_LOGIN_URL.rstrip('/')}{api_path('Auth', 'EPM', 'Logon')}" body = { 'Username': EPM_USERNAME, 'Password': EPM_PASSWORD, 'ApplicationID': EPM_APPLICATION_ID, } data = request_json('POST', url, {'Content-Type': 'application/json'}, body) token = data.get('EPMAuthenticationResult') manager_url = data.get('ManagerURL') if not token or not manager_url: raise FetchError(f'Login response did not contain a token and manager URL: {data}') if data.get('IsPasswordExpired'): raise FetchError('The EPM user password has expired; set a new password and update EPM_PASSWORD') return manager_url.rstrip('/'), token def list_sets(manager_url: str, headers: dict): """Return the sets visible to the API user.""" url = f"{manager_url}{api_path('Sets')}?{urlencode({'Offset': 0, 'Limit': 1000})}" data = request_json('GET', url, headers) sets = data.get('Sets') or [] print(f"Found {len(sets)} sets") return sets def fetch_events(manager_url: str, headers: dict, set_id: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int): """Fetch raw events for one set from the EPM Events/Search API. Pages with nextCursor, which returns events in chronological order. A time watermark is only correct over an ascending scan: with newest-first ordering a truncated run would advance the watermark past unread events. Returns: Tuple of (records list, newest_event_time ISO string). """ endpoint = f"{manager_url}{api_path('Sets', set_id, 'Events', 'Search')}" body = { 'filter': f'eventDate GE {epm_time(start_time)} AND eventDate LE {epm_time(end_time)}', } records = [] newest_time = None cursor = 'start' page_num = 0 while True: page_num += 1 if len(records) >= max_records: # Stop cleanly. The remaining events stay ahead of the watermark and # are collected by the next run. print(f"Set {set_id}: reached max_records limit ({max_records})") break # Rebuild the full query on every page; the time window must never be dropped. url = f"{endpoint}?{urlencode({'nextCursor': cursor, 'limit': page_size})}" data = request_json('POST', url, headers, body) page_results = data.get('events') or [] if not page_results: print(f"Set {set_id}: no more results (empty page)") break print(f"Set {set_id}: page {page_num} retrieved {len(page_results)} events") records.extend(page_results) for event in page_results: stamp = event_time(event) if not stamp: continue try: if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time): newest_time = stamp except ValueError as e: print(f"Warning: Could not parse event time {stamp!r}: {e}") cursor = data.get('nextCursor') if not cursor: break print(f"Set {set_id}: retrieved {len(records)} total events from {page_num} pages") return records, newest_time第二個檔案 - requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
按一下「部署」，即可儲存並部署函式。
等待部署作業完成 (2 至 3 分鐘)。
函式如何收集事件
- 這個函式會在每次執行時登入一次，列出使用者可查看的資料集，並依時間順序呼叫每個資料集的「取得詳細原始事件」API，並對
eventDate進行篩選，以及使用nextCursor分頁。 - 每個事件都會寫入額外的
setID欄位，以便將集合保留在記錄中。 - 狀態檔案會針對每個集合，保留最新看到的
lastEventDate，以及重疊視窗內事件的雜湊。每次執行作業時，系統都會重新讀取該浮水印之前的OVERLAP_MINUTES，並捨棄已寫入的事件，因此即使代理程式延遲回報，系統仍會收集資料，不會重複。 - 如果發生任何 API、登入或儲存空間錯誤，執行作業就會失敗，浮水印也不會變更，因此下次執行作業時會重試相同的時間範圍。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。
- 前往 Google Cloud 控制台的「Cloud Scheduler」。
- 點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 cyberark-epm-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議選取世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 cyberark-epm-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間需求選擇頻率：
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 15 分鐘 | */15 * * * * |
低延遲 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
低量、批次處理 |
測試整合項目
- 在 Cloud Scheduler 控制台中，找出您的工作。
- 按一下「強制執行」，手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「服務」。
- 按一下函式名稱 (
cyberark-epm-collector)。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。尋找：
Found N sets Set <set-id> (<set-name>): fetching events from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Set <set-id>: page 1 retrieved X events Wrote X records to gs://cyberark-epm-secops-logs/cyberark-epm/events_YYYYMMDDTHHMMSSZ.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往前置字元資料夾 (
cyberark-epm/)。確認是否已建立含有目前時間戳記的新
.ndjson檔案。
如果記錄中顯示錯誤：
- HTTP 401：請檢查
EPM_USERNAME和EPM_PASSWORD，或確認是否在一分鐘內重複登入 - HTTP 403：使用者在集合中沒有角色，或是租戶的 API 用量限制已達上限
- HTTP 404：
EPM_SET_IDS中的設定 ID 錯誤 - HTTP 429：頻率限制 - 函式會以退避演算法重試
- 缺少環境變數：檢查是否已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 Cloud Storage bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
CyberArk EPM logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「CyberArk Endpoint Privilege Manager (EPM)」做為「記錄類型」。
- 按一下「取得服務帳戶」。
系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 Cloud Storage bucket URI，並加上前置路徑：
gs://cyberark-epm-secops-logs/cyberark-epm/- 將
cyberark-epm-secops-logs替換為 Cloud Storage bucket 名稱。
- 將
來源刪除選項：根據偏好設定選取刪除選項：
- 一律不刪除檔案：一律不刪除來源中的檔案 (建議用於測試)。
刪除已轉移的檔案和空白目錄：成功擷取檔案後，從來源刪除檔案和空白目錄。
檔案存在時間上限：納入在過去天數內修改的檔案 (預設為 180 天)
資產命名空間：資產命名空間
擷取標籤：要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的新聞動態消息設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩項角色：Storage 物件檢視者 (可讀取記錄物件)，以及 bucket 層級角色 (可讀取 bucket 中繼資料)。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色：選取下列兩個角色：
- Storage 物件檢視者：讀取記錄物件。
- Storage 舊版 bucket 讀取者：讀取 bucket 中繼資料。如果您選取「刪除已轉移的檔案和空白目錄」刪除選項，請改為選取「Storage Legacy Bucket Writer」，這也會授予刪除權限。
- 按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
accessTargetType_label |
additional.fields |
已合併 |
accessaction_label |
additional.fields |
已合併 |
agentEventCount_label |
additional.fields |
已合併 |
arrivalTime_label |
additional.fields |
已合併 |
deceptionType_label |
additional.fields |
已合併 |
eventCount_label |
additional.fields |
已合併 |
logonAttemptTypeId_label |
additional.fields |
已合併 |
logonStatusId_label |
additional.fields |
已合併 |
modificationTime_label |
additional.fields |
已合併 |
originalFileName_label |
additional.fields |
已合併 |
owner_label |
additional.fields |
已合併 |
packageName_label |
additional.fields |
已合併 |
publisher_field |
additional.fields |
已合併 |
publisher_label |
additional.fields |
已合併 |
setID_label |
additional.fields |
已合併 |
skippedCount_label |
additional.fields |
已合併 |
sourceProcessSigner_field |
additional.fields |
已合併 |
sourceProcessSigner_label |
additional.fields |
已合併 |
sourceType_label |
additional.fields |
已合併 |
threatProtectionActionId_label |
additional.fields |
已合併 |
threatProtectionAction_label |
additional.fields |
已合併 |
user_is_admin_label |
additional.fields |
已合併 |
winEventRecordId_label |
additional.fields |
已合併 |
winEventType_label |
additional.fields |
已合併 |
workingDirectory_label |
additional.fields |
已合併 |
displayName |
metadata.description |
直接對應 |
eachevent.displayName |
metadata.description |
直接對應 |
lastEventDate |
metadata.event_timestamp |
剖析為 ISO8601 |
has_principal |
metadata.event_type |
已對應：true → STATUS_UPDATE |
has_principal_user |
metadata.event_type |
已對應：true → USER_UNCATEGORIZED |
eachevent.eventType |
metadata.product_event_type |
直接對應 |
eventType |
metadata.product_event_type |
直接對應 |
productName |
metadata.product_version |
直接對應 |
productVersion |
metadata.product_version |
直接對應 |
url |
network.http.referral_url |
直接對應 |
agentId |
principal.asset.asset_id |
直接對應 |
eachevent.agentId |
principal.asset.asset_id |
直接對應 |
computerName |
principal.asset.hostname |
直接對應 |
sourceName |
principal.file.full_path |
直接對應 |
computerName |
principal.hostname |
直接對應 |
eachevent.computerName |
principal.hostname |
直接對應 |
operatingSystemType |
principal.platform |
直接對應 |
operating_system_type |
principal.platform |
已對應：Windows → WINDOWS |
company |
principal.user.company_name |
直接對應 |
eachevent.userName |
principal.user.userid |
直接對應 |
userName |
principal.user.userid |
直接對應 |
sec_result |
security_result |
已合併 |
threat_protection_action |
security_result.action_details |
直接對應 |
fileDescription |
security_result.description |
直接對應 |
eachevent.policyName |
security_result.rule_name |
直接對應 |
policyName |
security_result.rule_name |
直接對應 |
justification |
security_result.summary |
直接對應 |
fileQualifier |
target.asset.asset_id |
直接對應 |
software |
target.asset.software |
已合併 |
eachevent.filePath |
target.file.full_path |
直接對應 |
filePath |
target.file.full_path |
直接對應 |
eachevent.hash |
target.file.sha1 |
直接對應 |
hash |
target.file.sha1 |
直接對應 |
fileSize |
target.file.size |
直接對應 |
eachevent.processCommandLine |
target.process.command_line |
已重新命名/對應 |
processCommandLine |
target.process.command_line |
直接對應 |
fileName |
target.process.file.names |
已合併 |
eachevent.sourceProcessCommandLine |
target.process.parent_process.command_line |
已重新命名/對應 |
sourceProcessCommandLine |
target.process.parent_process.command_line |
直接對應 |
each_event.sourceProcessHash |
target.process.parent_process.file.sha1 |
已重新命名/對應 |
sourceProcessHash |
target.process.parent_process.file.sha1 |
直接對應 |
sourceProcessPublisher_label |
target.resource.attribute.labels |
已合併 |
exposed_accountName |
target.user.attribute.labels |
已合併 |
exposed_domain |
target.user.attribute.labels |
已合併 |
exposed_source |
target.user.attribute.labels |
已合併 |
exposed_username |
target.user.attribute.labels |
已合併 |
sourceProcessUsername |
target.user.userid |
直接對應 |
| 不適用 | metadata.event_type |
常數：STATUS_UPDATE |
| 不適用 | metadata.product_name |
常數：EPM |
| 不適用 | metadata.vendor_name |
常數：CYBERARK |
| 不適用 | principal.platform |
常數：WINDOWS |
| 不適用 | security_result.alert_state |
常數：ALERTING |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。