收集 Proofpoint TAP Threats 日志
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Cloud Storage V2 将 Proofpoint TAP Threats 日志注入到 Google Security Operations。Proofpoint Targeted Attack Protection (TAP) 是一种先进的电子邮件安全平台，可检测、分析和阻止通过电子邮件传递的威胁，包括恶意附件和网址。TAP Threat API 会返回 TAP 在您的环境中发现的每种威胁的摘要：其分类、状态、严重程度、检测类型，以及与之关联的威胁主体、恶意软件系列、技术和品牌。Cloud Run 函数通过 TAP SIEM API 发现环境中观察到的威胁，并将其 Threat API 摘要写入 Cloud Storage 存储桶，Google SecOps 会从该存储桶中提取这些摘要。
准备工作
请确保满足以下前提条件：
- Google SecOps 实例
- 启用了 Cloud Storage API 的 Google Cloud 项目
- 创建和管理 Cloud Storage 存储桶的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 具有访问“威胁洞察信息中心”权限的 Proofpoint TAP 订阅
- 具有 SIEM API 访问权限的 TAP API 服务凭据（服务主体和密钥）
生成 Proofpoint TAP API 服务凭证
- 登录 Proofpoint TAP Threat Insight Dashboard。
- 依次前往设置 > 已关联的应用 > 服务凭据。
- 点击创建新凭证。
在生成的服务凭证对话框中，复制并安全地存储以下内容：
- 服务主体：用于 API 身份验证的主体标识符
- 密钥：用于 API 身份验证的密钥
验证 API 访问权限
在继续进行集成之前，请按以下步骤测试您的凭据：
PRINCIPAL="your-service-principal" SECRET="your-secret" # Test SIEM API access (fetch last hour of events) curl -s "https://tap-api-v2.proofpoint.com/v2/siem/all?format=json&sinceSeconds=3600" \ --user "${PRINCIPAL}:${SECRET}"如果响应成功，则会返回一个 JSON 对象，其中包含
messagesBlocked、messagesDelivered、clicksBlocked和clicksPermitted数组。如果您收到401错误，请验证您的服务主体和密钥是否正确。如果您收到403错误，请确认您的账号已启用 TAP API 访问权限。然后，从该响应中的任意事件中获取
threatID值，并测试 Threat API，该 API 是本指南所提取记录的来源：THREAT_ID="<threatID from a SIEM event>" curl -s "https://tap-api-v2.proofpoint.com/v2/threat/summary/${THREAT_ID}" \ --user "${PRINCIPAL}:${SECRET}"成功的响应是一个 JSON 对象，其中包含
id、identifiedAt、name、type、category、status以及actors、families、malware、techniques和brands数组。如果 SIEM 响应中没有事件，则表示环境在过去一小时内没有出现 TAP 识别的威胁；该函数仍会正常运行，但在出现威胁之前不会写入任何内容。
创建 Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建一个新项目。
- 在导航菜单中，前往 Cloud Storage > 存储桶。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 指定存储桶的名称 输入一个全局唯一的名称（例如 proofpoint-tap-threats-logs）位置类型 根据您的需求进行选择（单区域、双区域、多区域） 位置 选择位置（例如 us-central1）存储类别 标准（建议用于频繁访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有写入 Cloud Storage 存储桶和被 Pub/Sub 调用的权限。
创建服务账号
- 在 GCP 控制台中，前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
tap-threats-collector-sa - 服务账号说明：输入
Service account for Cloud Run function to collect Proofpoint TAP Threats 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 存储桶的 Identity and Access Management (IAM) 权限
向服务账号授予对 Cloud Storage 存储桶的写入权限：
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
tap-threats-collector-sa@PROJECT_ID.iam.gserviceaccount.com） - 分配角色：选择 Storage Object Admin
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
tap-threats-collector-trigger - 将其他设置保留为默认值
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集威胁事件
- Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发。它会读取 TAP SIEM API 以查找在您的环境中发现的威胁 ID，从 TAP Threat API 中提取每种威胁的摘要，并将摘要写入 Cloud Storage。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击编写函数。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 tap-threats-collector区域 选择与您的 Cloud Storage 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器部分中，点击添加触发器，然后选择 Pub/Sub 触发器。
在 Eventarc 触发器窗格中，提供以下配置详细信息：
- 触发器名称：保留生成的名称，或输入触发器的名称。
- 触发器类型：选择 Google 来源。
- 事件提供方：选择 Pub/Sub。
- 事件类型：选择 google.cloud.pubsub.topic.v1.messagePublished。
- 选择 Cloud Pub/Sub 主题：选择主题
tap-threats-collector-trigger。 - 区域：选择与函数相同的区域。
- 服务账号：选择服务账号
tap-threats-collector-sa。
点击保存触发器。
在身份验证部分中：
- 选择需要进行身份验证。
- 选择 Identity and Access Management (IAM)。
前往并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号
tap-threats-collector-sa
- 服务账号：选择服务账号
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量：
变量名称 示例值 说明 GCS_BUCKETproofpoint-tap-threats-logsCloud Storage 存储桶名称 GCS_PREFIXtap-threats日志文件的前缀 STATE_KEYtap-threats-state.json状态路径，位于日志前缀之外 TAP_PRINCIPALyour-service-principalTAP API Service Principal TAP_SECRETyour-secretTAP API 密钥 LOOKBACK_HOURS24首次运行时的初始发现窗口（以小时为单位，最大值为 168，即 SIEM API 限制） OVERLAP_MINUTES10重新读取水印之前的时间窗口，以捕获延迟编入索引的 SIEM 事件 MAX_THREATS500每次运行提取的威胁摘要数量上限；其余摘要将保留在状态文件中排队 SEEN_RETENTION_DAYS7威胁 ID 在上次被发现后被记住的天数 在变量和密钥部分中，前往请求：
- 请求超时：输入
540秒（9 分钟）
- 请求超时：输入
前往设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值
- CPU：选择 1
- 在资源部分中：
在修订版本伸缩部分中：
- 实例数下限：输入
0 - 实例数上限：输入
100（或根据预期负载进行调整）
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在入口点字段中输入 main。
在内嵌代码编辑器中，创建两个文件：
- 第一个文件 - main.py:：
import functions_framework from google.cloud import storage from google.cloud.exceptions import NotFound import base64 import hashlib import json import os import urllib3 from urllib.parse import quote 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', 'tap-threats') # 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', 'tap-threats-state.json') API_BASE = os.environ.get('TAP_API_BASE', 'https://tap-api-v2.proofpoint.com') TAP_PRINCIPAL = os.environ.get('TAP_PRINCIPAL') TAP_SECRET = os.environ.get('TAP_SECRET') # The SIEM API accepts at most 7 days of lookback. LOOKBACK_HOURS = min(int(os.environ.get('LOOKBACK_HOURS', '24')), 168) # The discovery query re-reads this many minutes before the watermark so that # events the SIEM API indexes late still surface their threat IDs. OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '10')) # Maximum threat summaries fetched per run. The rest stay queued in the state. MAX_THREATS = int(os.environ.get('MAX_THREATS', '500')) # Days a threat ID stays in the state after it was last seen in SIEM events. SEEN_RETENTION_DAYS = int(os.environ.get('SEEN_RETENTION_DAYS', '7')) class FetchError(Exception): """Raised when a Proofpoint API call fails. The watermark must never advance on a failed fetch, otherwise the threats referenced in the failed window are never looked up. """ 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 tap_time(dt: datetime) -> str: """Render a datetime in the YYYY-MM-DDThh:mm:ssZ form the SIEM API accepts.""" return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') def auth_header() -> str: """Build the HTTP Basic header from the service principal and secret.""" token = base64.b64encode(f'{TAP_PRINCIPAL}:{TAP_SECRET}'.encode('utf-8')).decode('utf-8') return f'Basic {token}' def content_hash(record: dict) -> str: """Hash a threat summary so an unchanged summary is not written twice.""" return hashlib.sha256( json.dumps(record, sort_keys=True, ensure_ascii=False).encode('utf-8') ).hexdigest() @functions_framework.cloud_event def main(cloud_event): """Discover threat IDs in TAP SIEM events and write their Threat API summaries. The PROOFPOINT_TAP_THREATS parser reads the Threat API summary schema, not the SIEM API events. The SIEM API is used only to find which threat IDs were observed; every message and click event for those threats is what the Proofpoint TAP (PROOFPOINT_MAIL) feed carries. Args: cloud_event: CloudEvent object containing the Pub/Sub message. """ if not all([GCS_BUCKET, TAP_PRINCIPAL, TAP_SECRET]): # 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) now = datetime.now(timezone.utc) watermark = None if state.get('last_event_time'): watermark = parse_datetime(state['last_event_time']) # threat_id -> {'hash': sha256 of the last written summary, 'seen': ISO time} seen = state.get('seen_threats', {}) pending = list(state.get('pending_threat_ids', [])) if watermark is None: start_time = now - timedelta(hours=LOOKBACK_HOURS) else: start_time = max(watermark - timedelta(minutes=OVERLAP_MINUTES), now - timedelta(days=7)) headers = { 'Authorization': auth_header(), 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-TAPThreatsCollector/2.0', } print(f"Discovering threats in SIEM 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. discovered, newest_event_time = discover_threat_ids(headers, start_time, now) for threat_id in discovered: seen.setdefault(threat_id, {})['seen'] = now.isoformat() if threat_id not in pending: pending.append(threat_id) print(f"Discovered {len(discovered)} threat IDs, {len(pending)} queued for lookup") batch = pending[:MAX_THREATS] fresh = [] for threat_id in batch: summary = fetch_threat_summary(headers, threat_id) if summary is None: # The Threat API has no summary for this ID; drop it from the queue. seen.setdefault(threat_id, {})['seen'] = now.isoformat() continue digest = content_hash(summary) if seen.get(threat_id, {}).get('hash') == digest: # Unchanged since it was last written; nothing new to ingest. continue fresh.append((threat_id, digest, summary)) remaining = pending[len(batch):] if remaining: print(f"Reached max_threats limit ({MAX_THREATS}); {len(remaining)} threat IDs stay queued") # Forget threat IDs not seen for SEEN_RETENTION_DAYS so the state stays small. cutoff = now - timedelta(days=SEEN_RETENTION_DAYS) seen = { tid: info for tid, info in seen.items() if info.get('seen') and parse_datetime(info['seen']) >= cutoff } if fresh: timestamp = now.strftime('%Y%m%dT%H%M%SZ') object_key = f"{GCS_PREFIX}/threats_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join(json.dumps(summary, ensure_ascii=False) for _, _, summary in fresh) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(fresh)} threat summaries to gs://{GCS_BUCKET}/{object_key}") for threat_id, digest, _ in fresh: seen.setdefault(threat_id, {})['hash'] = digest else: print("No new or changed threat summaries.") # Advance the watermark only after the data is durably written. When the # window held no events at all, keep the old watermark so a late-indexed # event is still picked up by the next overlap. new_watermark = watermark if newest_event_time: candidate = parse_datetime(newest_event_time) if new_watermark is None or candidate > new_watermark: new_watermark = candidate save_state(bucket, STATE_KEY, { 'last_event_time': new_watermark.isoformat() if new_watermark else None, 'seen_threats': seen, 'pending_threat_ids': remaining, }) print(f"Successfully processed {len(fresh)} threat summaries") 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: last_event_time={state.get('last_event_time')}, " f"{len(state.get('seen_threats', {}))} known threats, " f"{len(state.get('pending_threat_ids', []))} pending") def request_json(url: str, headers: dict): """Call a TAP API endpoint and return the decoded JSON body. Returns None on HTTP 404, which the Threat API uses for an unknown ID. Raises FetchError on any other transport or HTTP failure, so that the caller does not mistake a failed call for an empty result. """ backoff = 1.0 for attempt in range(5): try: response = http.request('GET', url, 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, 60.0) continue if response.status == 404: return None if response.status != 200: raise FetchError(f'HTTP {response.status} from {url}: {response.data.decode("utf-8", "replace")[:500]}') 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 discover_threat_ids(headers: dict, start_time: datetime, end_time: datetime): """Read TAP SIEM events in one-hour windows and collect the threat IDs they reference. Returns: Tuple of (ordered list of unique threat IDs, newest event time ISO string). """ threat_ids = [] newest_time = None current_start = start_time while current_start < end_time: # The SIEM API allows at most one hour per request. current_end = min(current_start + timedelta(hours=1), end_time) interval = f'{tap_time(current_start)}/{tap_time(current_end)}' url = f'{API_BASE}/v2/siem/all?format=json&interval={quote(interval, safe="/:")}' data = request_json(url, headers) if data is None: raise FetchError(f'SIEM API returned 404 for interval {interval}') window_count = 0 for key in ('messagesBlocked', 'messagesDelivered'): for message in data.get(key) or []: window_count += 1 stamp = message.get('messageTime') for info in message.get('threatsInfoMap') or []: tid = info.get('threatID') if tid and tid not in threat_ids: threat_ids.append(tid) stamp = stamp or info.get('threatTime') if stamp and (newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time)): newest_time = stamp for key in ('clicksBlocked', 'clicksPermitted'): for click in data.get(key) or []: window_count += 1 tid = click.get('threatID') if tid and tid not in threat_ids: threat_ids.append(tid) stamp = click.get('clickTime') or click.get('threatTime') if stamp and (newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time)): newest_time = stamp print(f"Interval {interval}: {window_count} events, {len(threat_ids)} unique threats so far") current_start = current_end return threat_ids, newest_time def fetch_threat_summary(headers: dict, threat_id: str): """Fetch one threat summary from the Threat API, or None if TAP has no summary for it.""" url = f'{API_BASE}/v2/threat/summary/{quote(threat_id, safe="")}' summary = request_json(url, headers) if summary is None: print(f"Threat {threat_id}: no summary available (404)") return None if 'id' not in summary: summary['id'] = threat_id return summary- 第二个文件 - requirements.txt:：
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署函数。
等待部署完成（2-3 分钟）。
函数如何收集威胁
- 该函数会从其水印开始，以一小时为间隔读取 TAP SIEM API
/v2/siem/all端点，并收集被屏蔽和已递送的邮件以及被屏蔽和允许的点击所引用的threatID值。系统不会写入 SIEM 事件本身，只会使用威胁 ID。 - 对于每个发现的威胁 ID，它会调用
/v2/threat/summary/{threatId}并将摘要写入为一条 NDJSON 记录。仅当总结的内容自上次撰写以来发生变化时，系统才会重新撰写总结，例如当威胁状态从active变为cleared时。 - 状态文件会将最新的 SIEM 事件时间作为水印、每个威胁的上次写入摘要的哈希值，以及运行达到
MAX_THREATS时仍在排队的威胁 ID。每次运行都会在水印之前重新读取OVERLAP_MINUTES，并首先处理队列，因此当运行达到上限或事件索引延迟时，不会丢失任何内容。 任何 API 或存储错误都会导致运行失败，并使水印保持不变，因此下次运行会重试同一时间窗口。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP 控制台中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 tap-threats-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时整点）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择主题 tap-threats-collector-trigger邮件正文 {}（空 JSON 对象）点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次：
| 频率 | Cron 表达式 | 使用场景 |
|---|---|---|
| 每隔 15 分钟 | */15 * * * * |
威胁众多的高流量环境 |
| 每小时 | 0 * * * * |
标准（推荐） |
| 每 6 小时 | 0 */6 * * * |
低容量环境 |
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业。
- 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击函数名称
tap-threats-collector。 - 点击日志标签页。
验证函数是否已成功执行。查找：
Discovering threats in SIEM events from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Interval .../...: X events, Y unique threats so far Discovered Y threat IDs, Y queued for lookup Wrote Z threat summaries to gs://proofpoint-tap-threats-logs/tap-threats/threats_YYYYMMDDTHHMMSSZ.ndjson Successfully processed Z threat summaries前往 Cloud Storage > 存储桶。
点击您的存储桶名称。
前往前缀文件夹
tap-threats/。验证是否已创建具有当前时间戳的新
.ndjson文件。
如果您在日志中看到错误，请执行以下操作：
- HTTP 401：检查环境变量中的 TAP_PRINCIPAL 和 TAP_SECRET。验证服务主体和密钥是否正确。
- HTTP 403：确认您的 TAP 账号已启用 API 访问权限。
- HTTP 429：速率限制 - 函数将自动重试并进行退避。请考虑降低时间表频率。
- 没有新的或更改的威胁摘要：如果在相应时间范围内未观察到任何威胁，或者每个观察到的威胁都已写入相同的内容，则属于正常情况。TAP 仅报告由 URL 防御或附件防御功能识别的威胁。
- 缺少环境变量：检查是否已设置所有必需的变量。
在 Google SecOps 中配置 Feed 以注入 Proofpoint TAP Threats 日志
- 前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Proofpoint TAP Threats）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Proofpoint Tap Threats 作为日志类型。
点击获取服务账号。系统会显示一个专属的服务账号邮箱，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 Cloud Storage 存储桶 URI：
gs://proofpoint-tap-threats-logs/tap-threats/- 替换：
proofpoint-tap-threats-logs：您的 Cloud Storage 存储桶名称。tap-threats：存储日志的可选前缀或文件夹路径（留空表示根目录）。
- 替换：
- 来源删除选项：根据您的偏好选择删除选项：
- 永不删除文件：永不从源中删除文件（建议用于测试）。
- 删除已转移的文件和空目录：在成功完成提取后，从来源位置删除文件和空目录。
- 文件存在时间上限：包含在过去指定天数内修改过的文件（默认值为 180 天）
- 资产命名空间：资产命名空间
- 注入标签：要应用于此 Feed 中事件的标签
点击下一步。
在最终确定界面中查看新的新闻信息流配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要对您的 Cloud Storage 存储桶拥有两个角色：用于读取日志对象的 Storage Object Viewer，以及用于读取存储桶元数据的存储桶级角色。
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
- 分配角色：选择以下两个角色：
- Storage Object Viewer：读取日志对象。
- Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储旧版存储桶写入者，该角色也会授予删除权限。
点击保存。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
detectionType_label |
additional.fields |
已合并 |
geoTargeted_label |
additional.fields |
已合并 |
notable_label |
additional.fields |
已合并 |
verticallyTargeted_label |
additional.fields |
已合并 |
identifiedAt |
metadata.event_timestamp |
已解析为 ISO8601 |
has_principal |
metadata.event_type |
已映射：true → NETWORK_CONNECTION，true → STATUS_UPDATE |
type |
metadata.product_event_type |
直接映射 |
id |
metadata.product_log_id |
直接映射 |
_associations |
security_result.associations |
已合并 |
_techniques |
security_result.attack_details.techniques |
已合并 |
_category |
security_result.category |
已合并 |
_category_details |
security_result.category_details |
已合并 |
notable |
security_result.priority |
已映射：true → HIGH_PRIORITY |
name |
security_result.threat_name |
直接映射 |
status |
security_result.threat_status |
已映射：active → ACTIVE，cleared → CLEARED |
_verdict_info |
security_result.verdict_info |
已合并 |
| 不适用 | metadata.event_type |
常量：NETWORK_CONNECTION |
| 不适用 | metadata.product_event_type |
常量：Proofpoint_Threats_Feed |
| 不适用 | metadata.product_name |
常量：TAP Threats |
| 不适用 | metadata.vendor_name |
常量：Proofpoint |
| 不适用 | security_result.priority |
常量：HIGH_PRIORITY |
| 不适用 | security_result.threat_status |
常量：ACTIVE |
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。