收集 Proofpoint Secure Email Relay 記錄

支援語言：

本文說明如何使用 Cloud Storage V2，將 Proofpoint Secure Email Relay 記錄檔擷取至 Google Security Operations。

Proofpoint Secure Email Relay (SER) 是一項受管理的轉發服務，適用於應用程式產生的郵件，例如交易和通知訊息。SER 會透過 Reporting API (SER 設定 API 之一) 產生該流量的報表。API 會傳回匯總用量數據：授權處理量位置、每日用量和訊息計數，以及每個中繼使用者的計數器。剖析器會將這些數字對應至統合式資料模型 (UDM)，並將這些數字記錄為標籤，與中繼使用者身分識別資訊一起記錄。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 Google Cloud 專案
  • 建立及管理 Cloud Storage bucket 的權限
  • 管理 Cloud Storage 值區身分與存取權管理 (IAM) 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • Proofpoint Secure Email Relay 入口網站的管理員存取權，包括建立 API 金鑰的權限

建立 Cloud Storage bucket

  1. 前往 Google Cloud 控制台。
  2. 選取專案或建立新專案。
  3. 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
  4. 按一下「建立值區」。

  5. 請提供下列設定詳細資料：

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 proofpoint-ser-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用：啟用物件版本管理或保留政策

  6. 點選「建立」。

收集 Proofpoint SER API 憑證

如要存取 Reporting API，您必須使用 Proofpoint API 金鑰管理服務核發的 API 金鑰。金鑰並非在 SER 控制台中建立，因此不會顯示在任何 SER 導覽項目下方。

建立 API 金鑰

  1. 使用管理員憑證登入 Proofpoint Secure Email Relay 入口網站。
  2. 開啟左上角的「應用程式切換器」。
  3. 依序前往「Services」(服務) >「API Key Management」(API 金鑰管理)。
  4. 按一下 [Create Key] (建立金鑰)。
  5. 選取「安全電子郵件轉送」做為產品。

  6. 複製並妥善保存下列值：

    • 金鑰：用做 client_id
    • 密鑰：用做 client_secret

驗證權限

API 金鑰管理服務會與多個 Proofpoint 產品共用，因此您必須透過產品切換工具存取，而非 SER 導覽選單。如要存取 SER 設定 API (包括 Reporting API)，請在建立金鑰時選取「Secure Email Relay」做為產品。

如果產品切換工具未顯示「API 金鑰管理」，表示您的帳戶沒有 API 管理權限。請與 Proofpoint 管理員或帳戶團隊聯絡，要求對方授予權限。

測試 API 存取權

  • Reporting API 使用 OAuth 2.0 用戶端憑證授權。將金鑰和密鑰換成不記名權杖，然後使用該權杖呼叫 API。

    # Replace with the values you copied
CLIENT_ID="<your-key>"
CLIENT_SECRET="<your-secret>"

# 1. Exchange the key and secret for an access token
ACCESS_TOKEN=$(curl -s -X POST "https://auth.proofpoint.com/v1/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=${CLIENT_ID}" \
  -d "client_secret=${CLIENT_SECRET}" \
  | python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 2. Call the Reporting API with the token
curl -v -H "Authorization: Bearer ${ACCESS_TOKEN}" \
  "https://reporting.ser.proofpoint.com/v1/usage/overview"

如果呼叫成功，系統會傳回 JSON 物件，其中的 data 成員包含 throughputLimit、licenseStartDate、licenseEndDate 和平均處理量。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備寫入 Cloud Storage bucket 權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

  1. 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
  2. 按一下「Create Service Account」(建立服務帳戶)。
  3. 請提供下列設定詳細資料：
    • 服務帳戶名稱：輸入 proofpoint-ser-collector-sa
    • 服務帳戶說明：輸入 Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs
  4. 按一下「建立並繼續」。
  5. 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
    1. 按一下「選擇角色」。
    2. 搜尋並選取「Storage 物件管理員」。
    3. 點選「+ 新增其他角色」。
    4. 搜尋並選取「Cloud Run Invoker」。
    5. 點選「+ 新增其他角色」。
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
  6. 按一下「繼續」。
  7. 按一下「完成」。

這些角色適用於：

  • Storage Object Admin：將記錄檔寫入 Cloud Storage bucket，並管理狀態檔案
  • Cloud Run 叫用者：允許 Pub/Sub 叫用函式
  • Cloud Functions 叫用者：允許函式叫用

授予 Cloud Storage bucket 的 IAM 權限

授予服務帳戶 Cloud Storage 值區的寫入權限：

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱 (例如 proofpoint-ser-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：輸入服務帳戶電子郵件地址 (例如 proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • 指派角色：選取「Storage 物件管理員」
  6. 按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

  1. 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
  2. 按一下「建立主題」。
  3. 請提供下列設定詳細資料：
    • 主題 ID：輸入 proofpoint-ser-trigger
    • 其他設定保留預設值
  4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 傳送的 Pub/Sub 訊息觸發，從 SER Reporting API 擷取用量報告，並將報告寫入 Cloud Storage。

  1. 前往 GCP Console 的「Cloud Run」。
  2. 按一下「Create service」(建立服務)。
  3. 選取「函式」 (使用內嵌編輯器建立函式)。

  4. 在「設定」部分，提供下列設定詳細資料：

    設定 值
    服務名稱 proofpoint-ser-collector
    區域 選取與 Cloud Storage bucket 相符的區域 (例如 us-central1)
    執行階段 選取 Python 3.12 以上版本

  5. 在「觸發條件 (選用)」部分：

    1. 按一下「+ 新增觸發條件」。
    2. 選取「Cloud Pub/Sub」。
    3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (proofpoint-ser-trigger)。
    4. 按一下 [儲存]。

  6. 在「Authentication」(驗證) 部分：

    1. 選取「需要驗證」。
    2. 檢查 Identity and Access Management (IAM)。

  7. 前往並展開「Containers, Networking, Security」。

  8. 前往「安全性」分頁：

    • 服務帳戶：選取服務帳戶 (proofpoint-ser-collector-sa)

  9. 前往「容器」分頁：

    1. 按一下「變數與密鑰」。
    2. 針對每個環境變數，按一下「+ 新增變數」：
    變數名稱 範例值 說明
    GCS_BUCKET proofpoint-ser-logs Cloud Storage Bucket 名稱
    GCS_PREFIX ser-logs 記錄檔的前置字串
    STATE_KEY ser-logs-state.json 狀態路徑 (記錄前置字元以外)
    TOKEN_URL https://auth.proofpoint.com/v1/token OAuth 2.0 權杖端點
    API_BASE https://reporting.ser.proofpoint.com SER Reporting API 基本網址
    CLIENT_ID your-key API 金鑰管理中的「金鑰」
    CLIENT_SECRET your-secret API 金鑰管理中的「Secret」
    PAGE_SIZE 50 每頁要求的 Relay 使用者數
    MAX_PAGES 200 轉送使用者分頁的頁面上限
    REPORT_DAY_OFFSET 1 回溯這麼多天後當天的報表
    MAX_BACKFILL_DAYS 7 單次執行最多可回填的天數
    SEEN_RETENTION_DAYS 7 狀態中保留簡化鍵的天數

  10. 在「變數與密鑰」部分中，前往「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)

  11. 前往「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上
      • CPU：選取 1

  12. 在「修訂版本資源調度」部分：

    • 執行個體數量下限：輸入 0
    • 執行個體數量上限：輸入 100 (或根據預期負載調整)

  13. 點選「建立」。

  14. 等待服務建立完成 (1 到 2 分鐘)。

  15. 服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

  1. 在「進入點」欄位中輸入「main」。
  2. 在內嵌程式碼編輯器中，建立兩個檔案：

  • 第一個檔案 - main.py:

    import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import time
import urllib.parse
import urllib3
from datetime import date, datetime, timezone, timedelta

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.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', 'ser-logs')
# 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', 'ser-logs-state.json')
TOKEN_URL = os.environ.get('TOKEN_URL', 'https://auth.proofpoint.com/v1/token')
API_BASE = os.environ.get('API_BASE', 'https://reporting.ser.proofpoint.com')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '50'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '200'))
# The Reporting API filters on whole calendar days, so the collector reports on
# a day that has already finished rather than on a partial one.
REPORT_DAY_OFFSET = int(os.environ.get('REPORT_DAY_OFFSET', '1'))
MAX_BACKFILL_DAYS = int(os.environ.get('MAX_BACKFILL_DAYS', '7'))
SEEN_RETENTION_DAYS = int(os.environ.get('SEEN_RETENTION_DAYS', '7'))

MAX_RATE_LIMIT_RETRIES = 5

class FetchError(Exception):
    """Raised when a Proofpoint SER API call fails.

    Collection must fail loudly. Returning an empty result on an API error is
    indistinguishable from a day with no traffic, and would let the run record
    a day it never actually read.
    """

def get_access_token():
    """Exchange the API key and secret for an OAuth 2.0 bearer token.

    The SER configuration APIs use the client credentials grant with the
    credentials in the form body. The returned token is sent to each API
    host as an Authorization: Bearer header.
    """
    body = urllib.parse.urlencode({
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
    })
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json',
    }

    try:
        response = http.request('POST', TOKEN_URL, body=body, headers=headers)
    except Exception as e:
        raise FetchError(f'Token request to {TOKEN_URL} failed: {e}') from e

    if response.status != 200:
        raise FetchError(f'HTTP {response.status} from {TOKEN_URL}: {response.data.decode("utf-8")}')

    try:
        token = json.loads(response.data.decode('utf-8')).get('access_token')
    except json.JSONDecodeError as e:
        raise FetchError(f'Malformed token response from {TOKEN_URL}: {e}') from e

    if not token:
        raise FetchError('Token response did not contain an access_token')

    return token

def call_api(token, method, path, body=None):
    """Call one Reporting API endpoint and return the decoded JSON body.

    Every failure raises. A caller that mistook an error for an empty report
    would record the day as collected and never come back to it.
    """
    url = f'{API_BASE}{path}'
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'User-Agent': 'GoogleSecOps-ProofpointSERCollector/1.0',
    }
    payload = json.dumps(body) if body is not None else None
    backoff = 1.0
    retries = 0

    while True:
        try:
            response = http.request(method, url, body=payload, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status == 429:
            retries += 1
            if retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError(f'Rate limited repeatedly by {url}; giving up without recording the day')
            raw_retry_after = response.headers.get('Retry-After')
            try:
                # Retry-After may also be an HTTP date, which int() cannot parse.
                delay = int(raw_retry_after) if raw_retry_after else int(backoff)
            except (TypeError, ValueError):
                delay = int(backoff)
            print(f'Rate limited (429) on {path}. Retrying after {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(f'HTTP {response.status} from {url}: {response.data.decode("utf-8")}')

        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

def as_list(value):
    """Return a response data member as a list.

    The Reporting API returns data as an object on some endpoints and as an
    array on others, and the published schema declares neither.
    """
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]

def collect_day(token, day, include_overview):
    """Collect one reporting day and return the records to write.

    The nesting of each record is chosen to match what the parser expects.
    Rows carrying acceptedMessages are emitted bare, because the parser nests
    those under data itself. The other two responses are emitted whole, so
    that their data and metadata envelope survives to the parser.
    """
    records = []

    # Licensed throughput position. This endpoint takes no date and reports
    # the tenant's standing right now, so it is collected once per run rather
    # than once per day: a backfill would otherwise write the same snapshot
    # against every day it catches up on.
    if include_overview:
        records.append(call_api(token, 'GET', '/v1/usage/overview'))

    # Volume in bytes for the day. One envelope per row, so that the
    # totalThroughput in metadata stays attached to the row it describes.
    data_trend = call_api(token, 'POST', '/v1/usage/data-trend', {
        'dates': day,
        'interval': 'day',
    })
    for element in as_list(data_trend.get('data')):
        records.append({'data': [element], 'metadata': data_trend.get('metadata', {})})

    # Message counters for the day.
    message_trend = call_api(token, 'POST', '/v1/usage/message-trend', {
        'dates': day,
        'interval': 'day',
    })
    records.extend(as_list(message_trend.get('data')))

    # Per relay user counters, paginated.
    page = 1
    while True:
        if page > MAX_PAGES:
            raise FetchError(f'Relay user pagination for {day} exceeded MAX_PAGES ({MAX_PAGES})')

        payload = call_api(token, 'POST', '/v1/usage/relay-users', {
            'dates': day,
            'pageNum': page,
            'pageSize': PAGE_SIZE,
        })
        elements = as_list(payload.get('data'))
        if not elements:
            break

        records.extend(elements)

        pagination = (payload.get('metadata') or {}).get('pagination') or {}
        total_pages = pagination.get('totalPages')
        if not total_pages or page >= total_pages:
            break
        page += 1

    return records

def record_key(record):
    """Return the deduplication identity of one record.

    Reporting rows are aggregates and carry no identifier of their own, so
    identity is the content hash. Keys are held per reporting day, which makes
    a repeated run of the same day a no-op without ever suppressing a new day
    whose figures happen to be identical.
    """
    return 'sha256:' + hashlib.sha256(
        json.dumps(record, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()

def pending_days(last_report_date, target):
    """Return the reporting days still to collect, oldest first."""
    if not last_report_date:
        start = target
    else:
        start = date.fromisoformat(last_report_date) + timedelta(days=1)

    if start > target:
        return []

    if (target - start).days >= MAX_BACKFILL_DAYS:
        start = target - timedelta(days=MAX_BACKFILL_DAYS - 1)
        print(f'Backfill capped at {MAX_BACKFILL_DAYS} days. Days before {start.isoformat()} are not collected.')

    return [start + timedelta(days=offset) for offset in range((target - start).days + 1)]

def prune_seen(seen, target):
    """Drop deduplication keys for days outside the retention window."""
    cutoff = target - timedelta(days=SEEN_RETENTION_DAYS)
    return {day: keys for day, keys in seen.items() if date.fromisoformat(day) >= cutoff}

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch Proofpoint SER usage reports and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, CLIENT_ID, CLIENT_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)
    seen = state.get('seen') or {}

    now = datetime.now(timezone.utc)
    target = (now - timedelta(days=REPORT_DAY_OFFSET)).date()

    days = pending_days(state.get('last_report_date'), target)
    if not days:
        print(f'Nothing to collect. {target.isoformat()} is already recorded.')
        return

    token = get_access_token()

    for day in days:
        key = day.isoformat()
        print(f'Collecting reporting day {key}')

        # A FetchError here propagates: the run fails, the state is left at the
        # last day that was fully written, and the next run retries this day.
        records = collect_day(token, key, include_overview=(day == days[-1]))

        day_seen = set(seen.get(key, []))
        fresh = []
        for record in records:
            digest = record_key(record)
            if digest in day_seen:
                continue
            day_seen.add(digest)
            fresh.append(record)

        print(f'{key}: fetched {len(records)} records, {len(fresh)} new after deduplication')

        if fresh:
            timestamp = now.strftime('%Y%m%dT%H%M%SZ')
            object_key = f'{GCS_PREFIX}/usage_{key}_{timestamp}.ndjson'
            blob = bucket.blob(object_key)

            ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
            blob.upload_from_string(ndjson, content_type='application/x-ndjson')

            print(f'Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}')

        # Record the day only after its data is durably written.
        seen[key] = sorted(day_seen)
        state['last_report_date'] = key
        state['seen'] = prune_seen(seen, target)
        save_state(bucket, STATE_KEY, state)

    print(f'Successfully processed {len(days)} reporting day(s)')

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 restart collection and re-ingest the
    whole backfill window.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the most recent completed day.')
        return {}

def save_state(bucket, key, state):
    """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 day 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_report_date={state.get("last_report_date")}')

  • 第二個檔案 - requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. 點選「部署」即可儲存並部署函式。

  2. 等待部署作業完成 (2 到 3 分鐘)。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。

  1. 前往 GCP 主控台的「Cloud Scheduler」。
  2. 點選「建立工作」。

  3. 請提供下列設定詳細資料：

    設定 值
    名稱 proofpoint-ser-collector-daily
    區域 選取與 Cloud Run 函式相同的區域
    頻率 0 2 * * * (每天 02:00)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取 Pub/Sub 主題 (proofpoint-ser-trigger)
    郵件內文 {} (空白 JSON 物件)

  4. 點選「建立」。

排程頻率選項

Reporting API 會依日曆天匯總資料，因此一天只會收集一次資料。在租戶時區的報表日期結束後，每天執行這項工作：

頻率 Cron 運算式 用途
每天凌晨 2 點 0 2 * * * 標準 (建議)
每天 06:00 0 6 * * * 報表結算時間較晚的房客
每天兩次 0 2,14 * * * 如果第一次執行失敗，會在當天重試

如果系統發現當天的資料已記錄，就會結束執行，不會寫入任何內容，因此額外執行一次只會耗用一次 API 呼叫，且絕不會重複資料。

測試整合項目

  1. 在 Cloud Scheduler 控制台中找出您的工作。
  2. 按一下「強制執行」即可手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「服務」。
  5. 按一下函式名稱 (proofpoint-ser-collector)。
  6. 點選「記錄」分頁標籤。

  7. 確認函式是否執行成功。尋找：

    Collecting reporting day YYYY-MM-DD
YYYY-MM-DD: fetched X records, X new after deduplication
Wrote X records to gs://proofpoint-ser-logs/ser-logs/usage_YYYY-MM-DD_YYYYMMDDTHHMMSSZ.ndjson
Saved state: last_report_date=YYYY-MM-DD
Successfully processed 1 reporting day(s)

  8. 依序前往「Cloud Storage」>「Buckets」。

  9. 按一下 bucket 名稱 (proofpoint-ser-logs)。

  10. 前往前置字元資料夾 (ser-logs/)。

  11. 確認是否已建立新的 .ndjson 檔案，且檔案名稱包含目前的時間戳記。

如果記錄中顯示錯誤：

  • 權杖端點傳回 HTTP 400：金鑰或密鑰有誤，或是金鑰已過期。invalid_client 表示金鑰組遭到拒絕，請建立替代金鑰。
  • Reporting API 傳回 HTTP 401：未傳送承載權杖或權杖已失效。
  • HTTP 429：速率限制。如果限制持續存在，函式會以退避演算法重試，並在執行失敗後將當天留給下一次執行。
  • 缺少環境變數：請檢查是否已設定所有必要變數。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 Cloud Storage bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Proofpoint SER Logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「ProofPoint Secure Email Relay」做為「記錄類型」。
  7. 按一下「取得服務帳戶」。

  8. 系統會顯示專屬服務帳戶電子郵件地址，例如：

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  9. 複製這個電子郵件地址，以便在下一步中使用。

  10. 點選「下一步」。

  11. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 Cloud Storage bucket URI，並加上前置路徑：

      gs://proofpoint-ser-logs/ser-logs/
      • 取代：
        • proofpoint-ser-logs：Cloud Storage bucket 名稱。
        • ser-logs：儲存記錄的選用前置字元或資料夾路徑 (如為根目錄，請留空)。

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 一律不刪除檔案：一律不刪除來源中的檔案 (建議用於測試)。
      • 刪除已轉移的檔案和空白目錄：成功擷取檔案後，從來源刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)

    • 資產命名空間：資產命名空間

    • 擷取標籤：要套用至這個動態饋給事件的標籤

  12. 點選「下一步」。

  13. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩個角色：Storage 物件檢視者，用於讀取記錄物件；以及 bucket 層級角色，用於讀取 bucket 中繼資料。

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱 (例如 proofpoint-ser-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址
    • 指派角色：選取下列兩個角色：
      • Storage 物件檢視者：讀取記錄檔物件。
      • Storage 舊版值區讀取者：讀取值區中繼資料。如果您選取「刪除已轉移的檔案和空目錄」刪除選項，請改為選取「Storage Legacy Bucket Writer」，這也會授予刪除權限。

  6. 按一下 [儲存]。

UDM 對應表

記錄欄位 UDM 對應 邏輯
acceptedMessages_label additional.fields 已合併
acceptedThroughput_label additional.fields 已合併
average30DayThroughput_label additional.fields 已合併
average7DayThroughput_label additional.fields 已合併
averageDailyThroughput_label additional.fields 已合併
avgAcceptedMessageSize_label additional.fields 已合併
blockedMessages_label additional.fields 已合併
data_totalThroughput_label additional.fields 已合併
deliveredMessages_label additional.fields 已合併
licenseEndDate_label additional.fields 已合併
licenseStartDate_label additional.fields 已合併
map_label additional.fields 已合併
metadata_totalThroughput_label additional.fields 已合併
quarantinedMessages_label additional.fields 已合併
rejectedMessages_label additional.fields 已合併
remainingThroughput_label additional.fields 已合併
requestedMessages_label additional.fields 已合併
requestedThroughput_label additional.fields 已合併
sentMessages_label additional.fields 已合併
throughputForecast_label additional.fields 已合併
throughputLimit_label additional.fields 已合併
throughput_label additional.fields 已合併
totalMessages_label additional.fields 已合併
undeliveredMessages_label additional.fields 已合併
data.name metadata.description 直接對應
desc metadata.description 直接對應
data.date metadata.event_timestamp 已剖析為 yyyy-MM-dd
event_type metadata.event_type 直接對應
fromEnvelope network.email.bounce_address 直接對應
fromHeader network.email.from 直接對應
applicationName principal.administrative_domain 直接對應
principal_host principal.asset.hostname 直接對應
principal_host principal.hostname 直接對應
principal_port principal.port 直接對應
data.relayUserId principal.user.product_object_id 直接對應
userId principal.user.product_object_id 直接對應
applicationUserName principal.user.user_display_name 直接對應
senderName target.administrative_domain 直接對應
senderId target.user.product_object_id 直接對應
不適用 metadata.product_name 常數：PROOFPOINT SER
不適用 metadata.vendor_name 常數：PROOFPOINT

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。