收集 Proofpoint Secure Email Relay 日志

支持的平台：

本文档介绍了如何使用 Cloud Storage V2 将 Proofpoint Secure Email Relay 日志注入到 Google Security Operations。

Proofpoint Secure Email Relay (SER) 是一种用于应用生成的邮件（例如交易和通知消息）的受管中继。SER 通过其报告 API（SER 配置 API 之一）报告该流量。该 API 会返回汇总的使用情况数据：许可的吞吐量位置、每日流量和消息数，以及每个中继用户的计数器。解析器会将这些数据映射到统一数据模型 (UDM)，并将其记录为中继用户身份旁边的标签。

准备工作

请确保满足以下前提条件：

  • Google SecOps 实例
  • 启用了 Cloud Storage API 的 Google Cloud 项目
  • 创建和管理 Cloud Storage 存储分区的权限
  • 管理 Cloud Storage 存储分区的 Identity and Access Management (IAM) 政策的权限
  • 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
  • 对 Proofpoint Secure Email Relay 门户的管理员访问权限，包括创建 API 密钥的权限

创建 Cloud Storage 存储桶

  1. 前往 Google Cloud 控制台。
  2. 选择您的项目或创建新项目。
  3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
  4. 点击创建存储分区。

  5. 提供以下配置详细信息：

    设置 值
    为存储桶命名 输入一个全局唯一的名称（例如 proofpoint-ser-logs）
    位置类型 根据您的需求进行选择（区域级、双区域、多区域）
    位置 选择营业地点（例如 us-central1）
    存储类别 标准（建议用于经常访问的日志）
    访问权限控制 均匀（推荐）
    保护工具 可选：启用对象版本控制或保留政策

  6. 点击创建。

收集 Proofpoint SER API 凭据

通过 Proofpoint API 密钥管理服务颁发的 API 密钥，可以访问 Reporting API。密钥不是在 SER 控制台本身中创建的，因此不会显示在任何 SER 导航条目下。

创建 API 密钥

  1. 使用管理员凭证登录 Proofpoint Secure Email Relay 门户。
  2. 打开左上角的应用切换器。
  3. 依次前往服务 > API 密钥管理。
  4. 点击创建密钥。
  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 存储桶的权限，并且可以由 Pub/Sub 调用。

创建服务账号

  1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
  2. 点击创建服务账号。
  3. 提供以下配置详细信息：
    • 服务账号名称：输入 proofpoint-ser-collector-sa
    • 服务账号说明：输入 Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs
  4. 点击创建并继续。
  5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
    1. 点击选择角色。
    2. 搜索并选择 Storage Object Admin。
    3. 点击 + 添加其他角色。
    4. 搜索并选择 Cloud Run Invoker。
    5. 点击 + 添加其他角色。
    6. 搜索并选择 Cloud Functions Invoker。
  6. 点击继续。
  7. 点击完成。

必须拥有这些角色，才能：

  • Storage Object Admin：将日志写入 Cloud Storage 存储桶并管理状态文件
  • Cloud Run Invoker：允许 Pub/Sub 调用函数
  • Cloud Functions Invoker：允许调用函数

授予对 Cloud Storage 存储桶的 IAM 权限

向服务账号授予对 Cloud Storage 存储桶的写入权限：

  1. 前往 Cloud Storage > 存储分区。
  2. 点击您的存储桶名称（例如 proofpoint-ser-logs）。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：输入服务账号电子邮件地址（例如 proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com）
    • 分配角色：选择 Storage Object Admin
  6. 点击保存。

创建 Pub/Sub 主题

创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。

  1. 在 GCP 控制台中，前往 Pub/Sub > 主题。
  2. 点击创建主题。
  3. 提供以下配置详细信息：
    • 主题 ID：输入 proofpoint-ser-trigger
    • 将其他设置保留为默认值
  4. 点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由 Cloud Scheduler 中的 Pub/Sub 消息触发，以从 SER Reporting API 中提取使用情况报告并将其写入 Cloud Storage。

  1. 在 GCP 控制台中，前往 Cloud Run。
  2. 点击创建服务。
  3. 选择函数（使用内嵌编辑器创建函数）。

  4. 在配置部分中，提供以下配置详细信息：

    设置 值
    Service 名称 proofpoint-ser-collector
    区域 选择与您的 Cloud Storage 存储桶匹配的区域（例如 us-central1）
    运行时 选择 Python 3.12 或更高版本

  5. 在触发器（可选）部分中：

    1. 点击 + 添加触发器。
    2. 选择 Cloud Pub/Sub。
    3. 在选择 Cloud Pub/Sub 主题中，选择 Pub/Sub 主题 (proofpoint-ser-trigger)。
    4. 点击保存。

  6. 在身份验证部分中：

    1. 选择需要进行身份验证。
    2. 检查 Identity and Access Management (IAM)。

  7. 前往并展开容器、网络、安全性。

  8. 前往安全性标签页：

    • 服务账号：选择服务账号 (proofpoint-ser-collector-sa)

  9. 前往容器标签页：

    1. 点击变量和密钥。
    2. 为每个环境变量点击+ 添加变量：
    变量名称 示例值 说明
    GCS_BUCKET proofpoint-ser-logs Cloud Storage 存储桶名称
    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 每页中继用户数
    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 Console 中，前往 Cloud Scheduler。
  2. 点击创建作业。

  3. 提供以下配置详细信息：

    设置 值
    名称 proofpoint-ser-collector-daily
    区域 选择与 Cloud Run 函数相同的区域
    频率 0 2 * * *（每天 02:00）
    时区 选择时区（建议选择 UTC）
    目标类型 Pub/Sub
    主题 选择 Pub/Sub 主题 (proofpoint-ser-trigger)
    消息正文 {}（空 JSON 对象）

  4. 点击创建。

时间表频率选项

Reporting API 按日历天进行汇总，因此完整的一天只收集一次。在租户的时区中报告日结束后，每天运行一次作业：

频率 Cron 表达式 使用场景
每天凌晨 2:00 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 > 存储分区。

  9. 点击您的存储桶名称 (proofpoint-ser-logs)。

  10. 前往前缀文件夹 (ser-logs/)。

  11. 验证是否已创建具有当前时间戳的新 .ndjson 文件。

如果您在日志中看到错误，请执行以下操作：

  • 来自令牌端点的 HTTP 400 错误：密钥或密钥错误，或者密钥已过期。invalid_client 表示相应密钥对已被拒绝；请创建替代密钥。
  • 来自 Reporting API 的 HTTP 401 错误：未发送不记名令牌或不记名令牌已失效。
  • HTTP 429：速率限制。该函数会进行退避重试，如果限制仍然存在，则运行失败，并将当天留给下一次运行。
  • 缺少环境变量：检查是否已设置所有必需的变量。

检索 Google SecOps 服务账号

Google SecOps 使用唯一的服务账号从您的 Cloud Storage 存储桶中读取数据。您必须向此服务账号授予对您的存储桶的访问权限。

获取服务账号电子邮件地址

  1. 依次前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 点击配置单个 Feed。
  4. 在 Feed 名称字段中，输入 Feed 的名称（例如 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. 为以下输入参数指定值：

    • 存储桶网址：输入带有前缀路径的 Cloud Storage 存储桶 URI：

      gs://proofpoint-ser-logs/ser-logs/
      • 替换：
        • proofpoint-ser-logs：您的 Cloud Storage 存储桶名称。
        • ser-logs：存储日志的可选前缀或文件夹路径（留空表示根目录）。

    • 来源删除选项：根据您的偏好选择删除选项：

      • 永不删除文件：永不从源中删除文件（建议用于测试）。
      • 删除已转移的文件和空目录：在成功提取完成后，从来源中删除文件和空目录。

    • 文件存在时间上限：包含在过去指定天数内修改过的文件（默认值为 180 天）

    • 资产命名空间：资产命名空间

    • 注入标签：要应用于此 Feed 中事件的标签

  12. 点击下一步。

  13. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

向 Google SecOps 服务账号授予 IAM 权限

Google SecOps 服务账号需要您的 Cloud Storage 存储桶具备两个角色：用于读取日志对象的 Storage Object Viewer，以及用于读取存储桶元数据的存储桶级角色。

  1. 前往 Cloud Storage > 存储分区。
  2. 点击您的存储桶名称（例如 proofpoint-ser-logs）。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
    • 分配角色：选择以下两个角色：
      • Storage Object Viewer：读取日志对象。
      • Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储空间旧版存储分区写入者，该角色也会授予删除权限。

  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 专业人士的解答。