Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Proofpoint Secure Email Relay のログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Cloud Storage V2 を使用して Proofpoint Secure Email Relay ログを Google Security Operations に取り込む方法について説明します。

Proofpoint Secure Email Relay（SER）は、トランザクションメッセージや通知メッセージなど、アプリケーションが生成するメールを管理するリレーです。SER は、SER 構成 API の 1 つである Reporting API を介して、そのトラフィックをレポートします。API は、ライセンスされたスループットの位置、1 日あたりのボリュームとメッセージ数、各リレーユーザーのカウンタなど、集計された使用状況の数値を返します。パーサーは、これらの数値を統合データモデル（UDM）にマッピングし、リレーユーザー ID とともにラベルとして記録します。

始める前に

次の前提条件を満たしていることを確認してください。

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 バケットを作成する

Google Cloud コンソールに移動します。
プロジェクトを選択するか、新しいプロジェクトを作成します。
ナビゲーションメニューで、[Cloud Storage] > [バケット] に移動します。
[バケットを作成] をクリックします。

次の構成情報を提供してください。

設定	値
バケットに名前を付ける	グローバルに一意の名前（`proofpoint-ser-logs` など）を入力します。
ロケーションタイプ	ニーズに応じて選択します（リージョン、デュアルリージョン、マルチリージョン）。
ロケーション	ロケーションを選択します（例: `us-central1`）。
ストレージクラス	Standard（頻繁にアクセスされるログに推奨）
アクセス制御	均一（推奨）
保護ツール	省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

[作成] をクリックします。

Proofpoint SER API 認証情報を収集する

Reporting API へのアクセスは、Proofpoint API 鍵管理サービスから発行された API キーによって許可されます。鍵は SER コンソール自体で作成されるわけではないため、SER のナビゲーションエントリには表示されません。

API キーを作成する

管理者認証情報を使用して Proofpoint Secure Email Relay ポータルにログインします。
左上のアプリスイッチャーを開きます。
[Services] > [API Key Management] に移動します。
[キーを作成] をクリックします。
プロダクトとして [Secure Email Relay] を選択します。
次の値をコピーして安全に保管します。
- キー: client_id として使用
- Secret: client_secret として使用されます。
注: キーとシークレットは画面に 1 回だけ表示されます。このシークレットは後で再度表示することはできないため、ページを離れる前に両方の値をコピーしてください。キーとシークレットは、有効化から 1 年後に期限切れになります。

権限を確認する

API Key Management サービスは複数の Proofpoint 製品で共有されているため、SER ナビゲーションメニューではなく、アプリスイッチャーからアクセスします。Reporting API を含む SER 構成 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"

呼び出しが成功すると、data メンバーに throughputLimit、licenseStartDate、licenseEndDate、スループットの平均を含む JSON オブジェクトが返されます。

Cloud Run functions のサービスアカウントを作成する

Cloud Run functions の関数には、Cloud Storage バケットへの書き込み権限と、Pub/Sub によって呼び出される権限を持つサービスアカウントが必要です。

サービスアカウントを作成する

GCP コンソールで、[IAM と管理] > [サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成情報を指定します。
- サービスアカウント名: 「proofpoint-ser-collector-sa」と入力します。
- サービスアカウントの説明: 「Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs」と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [Storage オブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [Cloud Functions 起動元] を検索して選択します。
[続行] をクリックします。
[完了] をクリックします。

これらのロールは、次の操作に必要です。

Storage オブジェクト管理者: Cloud Storage バケットにログを書き込み、状態ファイルを管理します。
Cloud Run 起動元: Pub/Sub が関数を呼び出せるようにします。
Cloud Functions 起動元: 関数の呼び出しを許可します。

Cloud Storage バケットに対する IAM 権限を付与する

Cloud Storage バケットに対する書き込み権限をサービスアカウントに付与します。

[Cloud Storage] > [バケット] に移動します。
バケット名（proofpoint-ser-logs など）をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成情報を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions の関数がサブスクライブする Pub/Sub トピックを作成します。

GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
[トピックを作成] をクリックします。
次の構成情報を指定します。
- トピック ID: 「proofpoint-ser-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、SER Reporting API から使用状況レポートを取得して Cloud Storage に書き込みます。

GCP Console で Cloud Run に移動します。
[サービスを作成] をクリックします。
[関数]（インライン編集を使用して関数を作成する）を選択します。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`proofpoint-ser-collector`
リージョン	Cloud Storage バケットに一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、Pub/Sub トピック（proofpoint-ser-trigger）を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数の呼び出し時に認証を自動的に処理します。
[コンテナ、ネットワーキング、セキュリティ] に移動して開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウント（proofpoint-ser-collector-sa）を選択します。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例	説明
`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 のベース URL
`CLIENT_ID`	`your-key`	API キー管理のキー
`CLIENT_SECRET`	`your-secret`	API キー管理のSecret
`PAGE_SIZE`	`50`	1 ページあたりのリクエストされたリレーユーザー数
`MAX_PAGES`	`200`	リレーユーザーのページネーションのページ上限
`REPORT_DAY_OFFSET`	`1`	指定した日数前のレポート
`MAX_BACKFILL_DAYS`	`7`	ほとんどの日で、1 回の実行でバックフィルが完了します。
`SEEN_RETENTION_DAYS`	`7`	重複除去キーが状態に保持される日数

[変数とシークレット] セクションで、[リクエスト] に移動します。
- リクエストタイムアウト: 600秒（10 分）を入力してください
[設定] タブに移動します。
- [リソース] セクションで、次の操作を行います。
  - メモリ: 512 MiB 以上を選択します。
  - CPU: 1 を選択します。
[リビジョンスケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数]: 「0」と入力します。
- [インスタンスの最大数]: 「100」と入力します（または、予想される負荷に基づいて調整します）。
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスが作成されると、インラインコードエディタが自動的に開きます。

関数のコードを追加する

[エントリポイント] フィールドに「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")}')

2 つ目のファイル - requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

[デプロイ] をクリックして関数を保存し、デプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。

注: Pub/Sub トリガー構成により、必要なサブスクリプションと権限が自動的に作成されます。

Cloud Scheduler ジョブを作成する

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run 関数をトリガーします。

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

次の構成情報を提供してください。

設定	値
名前	`proofpoint-ser-collector-daily`
リージョン	Cloud Run functions の関数と同じリージョンを選択します。
周波数	`0 2 * * *`（毎日、午前 2 時）
タイムゾーン	タイムゾーンを選択（UTC を推奨）
ターゲットタイプ	Pub/Sub
トピック	Pub/Sub トピック（`proofpoint-ser-trigger`）を選択します。
メッセージ本文	`{}`（空の JSON オブジェクト）

[作成] をクリックします。

スケジュール頻度のオプション

Reporting API は暦日ごとに集計するため、完了した日は 1 回収集されます。テナントのタイムゾーンでレポート対象日が終了した後に、ジョブを毎日実行します。

頻度	CRON 式	ユースケース
毎日 02:00	`0 2 * * *`	標準（推奨）
毎日 06:00	`0 6 * * *`	レポートの決済が遅いテナント
1 日 2 回	`0 2,14 * * *`	最初の実行が失敗した場合に同日の再試行を追加

すでに記録されている日を見つけた実行は、何も書き込まずに終了します。そのため、追加の実行には 1 回の API 呼び出しの費用がかかり、データが重複することはありません。

統合をテストする

Cloud Scheduler コンソールで、ジョブを見つけます。
[強制実行] をクリックして、ジョブを手動でトリガーします。
数秒待ちます。
[Cloud Run] > [サービス] に移動します。
関数名（proofpoint-ser-collector）をクリックします。
[ログ] タブをクリックします。

関数が正常に実行されたことを確認します。以下のものを探します。

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)

[Cloud Storage] > [バケット] に移動します。
バケット名（proofpoint-ser-logs）をクリックします。
接頭辞フォルダ（ser-logs/）に移動します。
現在のタイムスタンプで新しい .ndjson ファイルが作成されたことを確認します。

ログにエラーが表示される場合:

トークンエンドポイントからの HTTP 400: キーまたはシークレットが間違っているか、キーの有効期限が切れています。invalid_client は、ペアが拒否されたことを意味します。置換鍵を作成します。
Reporting API からの HTTP 401: 署名なしトークンが送信されていないか、有効期限が切れています。
HTTP 429: レート制限。関数はバックオフを使用して再試行し、制限が続く場合は実行を失敗させ、次の実行に持ち越します。
環境変数が設定されていない: 必要な変数がすべて設定されていることを確認します。

Google SecOps サービスアカウントを取得する

Google SecOps は、一意のサービスアカウントを使用して Cloud Storage バケットからデータを読み取ります。このサービスアカウントにバケットへのアクセス権を付与する必要があります。

サービスアカウントのメールアドレスを取得する

[設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Proofpoint SER Logs）。
[Source type] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [ProofPoint Secure Email Relay] を選択します。
[サービスアカウントを取得する] をクリックします。
一意のサービスアカウントのメールアドレスが表示されます。例:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

注: Google SecOps インスタンスごとに一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む Cloud Storage バケット URI を入力します。
```
gs://proofpoint-ser-logs/ser-logs/
```
  - 次のように置き換えます。
    - proofpoint-ser-logs: Cloud Storage バケット名。
    - ser-logs: ログが保存されるオプションの接頭辞またはフォルダパス（ルートの場合は空のままにします）。
注: URI の末尾には必ずスラッシュ（/）を付けてください。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - ファイルを削除しない: ソースからファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルと空のディレクトリを削除する: 取得が正常に完了した後、ソースからファイルと空のディレクトリを削除します。
- ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます（デフォルトは 180 日）。
- Asset namespace: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

Google SecOps サービスアカウントに IAM 権限を付与する

Google SecOps サービスアカウントには、Cloud Storage バケットに対する 2 つのロールが必要です。ログオブジェクトを読み取るためのストレージオブジェクト閲覧者ロールと、バケットメタデータを読み取るためのバケットレベルのロールです。

[Cloud Storage] > [バケット] に移動します。
バケット名（proofpoint-ser-logs など）をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: 次の両方を選択します。
  - Storage オブジェクト閲覧者: ログオブジェクトを読み取ります。
  - Storage レガシーバケット読み取り: バケットメタデータを読み取ります。削除オプションとして [転送されたファイルと空のディレクトリを削除する] を選択した場合は、代わりに [ストレージレガシーバケット書き込み] を選択します。これにより、削除権限も付与されます。
[保存] をクリックします。

注: 両方のロールを付与します。Storage オブジェクト閲覧者には、フィードに必要なバケットレベルの権限が付与されていません。また、Storage レガシーバケットのロールには、オブジェクトの内容を読み取る権限が付与されていません。

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 のプロフェッショナルから回答を得ることができます。