收集 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 存储桶，Google SecOps 会从中注入这些事件。
准备工作
请确保满足以下前提条件：
- Google SecOps 实例
- 启用了 Cloud Storage API 的 Google Cloud 项目
- 创建和管理 Cloud Storage 存储桶的权限
- 管理 Cloud Storage 存储桶的 Identity and Access Management (IAM) 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- CyberArk EPM SaaS 租户
- EPM 账号管理员，用于创建 API 用户和角色
创建 Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建一个新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储桶。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 cyberark-epm-secops-logs）位置类型 根据您的需求进行选择（区域级、双区域级、多区域级） 位置 选择位置（例如 us-central1）存储类别 标准（建议用于频繁访问的日志） 访问权限控制 统一（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
收集 CyberArk EPM 凭据
Cloud Run functions 函数使用 EPM 用户登录 EPM REST API。创建一个专用用户，该用户对您要收集其事件的媒体资源组具有“只能查看”权限。
创建 API 用户
- 以账号管理员身份登录 EPM 管理控制台。
- 点击管理板块。
- 前往用户页面，然后点击添加用户。
- 提供以下配置详细信息：
- 电子邮件地址：输入集成用户的登录名（例如
secops-collector@example.com）。 - 密码：输入 12 到 128 个字符的密码。
- 电子邮件地址：输入集成用户的登录名（例如
- 点击 Add（添加）。
- 妥善保存登录名和密码。您将使用它们作为 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 存储桶和被 Pub/Sub 调用的权限。
创建服务账号
- 在 Google Cloud 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
cyberark-epm-collector-sa - 服务账号说明：输入
Service account for Cloud Run function to collect CyberArk EPM logs
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
以下操作需要这些角色：
- Storage Object Admin：将日志写入 Cloud Storage 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 Cloud Storage 存储桶的 IAM 权限
向服务账号授予 Cloud Storage 存储桶的写入权限：
- 在 Google Cloud 控制台中，前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
cyberark-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com） - 分配角色：选择 Storage Object Admin
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建 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。
- 点击编写函数。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 cyberark-epm-collector区域 选择与您的 Cloud Storage 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器部分中，点击添加触发器，然后选择 Pub/Sub 触发器。
在 Eventarc 触发器窗格中，提供以下配置详细信息：
- 触发器名称：保留生成的名称，或输入触发器的名称。
- 触发器类型：选择 Google 来源。
- 事件提供方：选择 Pub/Sub。
- 事件类型：选择 google.cloud.pubsub.topic.v1.messagePublished。
- 选择 Cloud Pub/Sub 主题：选择主题
cyberark-epm-trigger。 - 区域：选择与函数相同的区域。
- 服务账号：选择服务账号
cyberark-epm-collector-sa。
点击保存触发器。
在身份验证部分中：
- 选择需要进行身份验证。
- 选择 Identity and Access Management (IAM)。
前往并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号
cyberark-epm-collector-sa
- 服务账号：选择服务账号
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量：
变量名称 示例值 说明 GCS_BUCKETcyberark-epm-secops-logsCloud Storage 存储桶名称 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 集；如果为空，则收集用户可以查看的每个集 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 分钟）。
该函数如何收集事件
- 该函数在每次运行时登录一次，列出用户可以查看的集合，并按时间顺序调用每个集合的 Get detailed raw events 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 * * * *（每小时整点）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 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 > 存储桶。
点击您的存储桶名称。
前往前缀文件夹 (
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 存储桶中读取数据。您必须向此服务账号授予对您的存储桶的访问权限。
获取服务账号邮箱
- 前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
CyberArk EPM logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 CyberArk Endpoint Privilege Manager (EPM) 作为日志类型。
- 点击获取服务账号。
系统会显示一个专属的服务账号邮箱，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 Cloud Storage 存储桶 URI：
gs://cyberark-epm-secops-logs/cyberark-epm/- 将
cyberark-epm-secops-logs替换为您的 Cloud Storage 存储桶名称。
- 将
来源删除选项：根据您的偏好选择删除选项：
- 永不删除文件：永不从来源中删除文件（建议在测试时使用）。
删除已转移的文件和空目录：在成功完成提取后，从来源位置删除文件和空目录。
文件存在时间上限：包含过去指定天数内修改过的文件（默认值为 180 天）
资源命名空间：资源命名空间
注入标签：要应用于此 Feed 中事件的标签
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要对您的 Cloud Storage 存储桶拥有两个角色：Storage Object Viewer（用于读取日志对象）和存储桶级角色（用于读取存储桶元数据）。
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
- 分配角色：选择以下两个角色：
- Storage Object Viewer：读取日志对象。
- Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储旧版存储桶写入者，该角色也会授予删除权限。
- 点击保存。
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 专业人士的解答。