收集 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 探索您環境中觀察到的威脅，並將威脅 API 摘要寫入 Cloud Storage bucket，供 Google SecOps 擷取。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 Google Cloud 專案
- 建立及管理 Cloud Storage bucket 的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 訂閱 Proofpoint TAP，並取得威脅洞察資訊主頁的存取權
- TAP API 服務認證 (服務主體和 Secret)，具備 SIEM API 存取權限
產生 Proofpoint TAP API 服務憑證
- 登入 Proofpoint TAP Threat Insight 資訊主頁。
- 依序前往「設定」>「已連結的應用程式」>「服務憑證」。
- 按一下「Create New Credential」(建立新憑證)。
在「產生的服務憑證」對話方塊中，複製並安全地儲存下列項目：
- 服務主體：用於 API 驗證的主體 ID
- 密鑰：用於 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}"如果回應成功，系統會傳回包含
messagesBlocked、messagesDelivered、clicksBlocked和clicksPermitted陣列的 JSON 物件。如果收到401錯誤，請確認服務主體和密碼正確無誤。如果收到403錯誤，請確認帳戶已啟用 TAP API 存取權。然後從該回應中的任一事件取得
threatID值，並測試 Threat 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 bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 proofpoint-tap-threats-logs)位置類型 請根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或資料保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要服務帳戶，且該帳戶須具備寫入 Cloud Storage bucket 的權限，並可由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
tap-threats-collector-sa - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Proofpoint TAP Threats logs
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- 儲存空間物件管理員：將威脅事件資料寫入 Cloud Storage bucket，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 Cloud Storage bucket 的身分與存取權管理 (IAM) 權限
授予服務帳戶 Cloud Storage bucket 的寫入權限：
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
tap-threats-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色：選取「Storage 物件管理員」
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 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」。
- 點選「編寫函式」。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 tap-threats-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 主題」：選擇主題
tap-threats-collector-trigger。 - 區域：選取與函式相同的區域。
- 服務帳戶：選取服務帳戶
tap-threats-collector-sa。
按一下「儲存觸發條件」。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 選取「Identity and Access Management (IAM)」。
前往並展開「容器、網路、安全性」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶
tap-threats-collector-sa
- 服務帳戶：選取服務帳戶
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETproofpoint-tap-threats-logsCloud Storage Bucket 名稱 GCS_PREFIXtap-threats記錄檔的前置字串 STATE_KEYtap-threats-state.json狀態路徑，位於記錄前置字元外 TAP_PRINCIPALyour-service-principalTAP API 服務主體 TAP_SECRETyour-secretTAP API 密鑰 LOOKBACK_HOURS24首次執行時的初始探索時間範圍 (以小時為單位，上限為 168，即 SIEM API 限制) OVERLAP_MINUTES10在浮水印前重新讀取視窗，擷取延遲建立索引的 SIEM 事件 MAX_THREATS500每次執行作業時擷取的威脅摘要數量上限；其餘摘要會保留在狀態檔案中，等待擷取 SEEN_RETENTION_DAYS7威脅 ID 在最後一次出現後保留的天數 在「變數與密鑰」部分，前往「要求」：
- 要求逾時：輸入
540秒 (9 分鐘)
- 要求逾時：輸入
前往「設定」分頁：
- 在「Resources」(資源) 部分：
- 記憶體：選取 512 MiB 以上
- 「CPU」：選取「1」
- 在「Resources」(資源) 部分：
在「Revision scaling」(修訂版本資源調度) 部分：
- 執行個體數量下限：輸入
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 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 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」>「Buckets」。
按一下 bucket 名稱。
前往前置字元資料夾
tap-threats/。確認是否已建立含有目前時間戳記的新
.ndjson檔案。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 TAP_PRINCIPAL 和 TAP_SECRET。確認服務主體和密鑰正確無誤。
- HTTP 403：確認 TAP 帳戶已啟用 API 存取權。
- HTTP 429：頻率限制 - 函式會自動重試，並延遲一段時間。建議您降低排程頻率。
- 沒有新增或變更的威脅摘要：如果在時間範圍內未觀察到任何威脅，或觀察到的威脅都已寫入相同內容，則這是正常現象。TAP 只會回報 URL Defense 或 Attachment Defense 識別出的威脅。
- 缺少環境變數：確認已設定所有必要變數。
在 Google SecOps 中設定動態饋給，擷取 Proofpoint TAP Threats 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Proofpoint TAP Threats)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Proofpoint Tap Threats」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 Cloud Storage bucket URI，並加上前置路徑：
gs://proofpoint-tap-threats-logs/tap-threats/- 取代：
proofpoint-tap-threats-logs：Cloud Storage bucket 名稱。tap-threats：選用前置字元或資料夾路徑，用於儲存記錄 (如為根目錄，請留空)。
- 取代：
- 來源刪除選項：根據偏好選取刪除選項：
- 一律不刪除檔案：一律不刪除來源中的檔案 (建議用於測試)。
- 刪除已轉移的檔案和空白目錄：成功擷取檔案後，從來源刪除檔案和空白目錄。
- 檔案存在時間上限：納入在過去天數內修改的檔案 (預設為 180 天)
- 資產命名空間：資產命名空間
- 擷取標籤：要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩項角色：Storage 物件檢視者 (可讀取記錄物件)，以及 bucket 層級角色 (可讀取 bucket 中繼資料)。
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址
- 「指派角色」：選取下列兩項：
- Storage 物件檢視者：讀取記錄物件。
- Storage 舊版 bucket 讀取者：讀取 bucket 中繼資料。如果您選取「刪除已轉移的檔案和空白目錄」刪除選項，請改為選取「Storage Legacy Bucket Writer」，這也會授予刪除權限。
按一下 [儲存]。
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 專家尋求解答。