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 とともにラベルとして記録します。

注: SER レポートは集計されます。API はメッセージごとのリソースを公開しないため、取り込まれたイベントには、件名、受信者、メッセージ 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 バケットを作成する

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

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

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

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

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

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

API キーを作成する

1. 管理者の認証情報を使用して Proofpoint Secure Email Relay ポータルにログインします。
2. 左上のアプリ スイッチャーを開きます。
3. [サービス] > [API キーの管理] に移動します。
4. [キーを作成] をクリックします。
5. プロダクトとして [Secure Email Relay] を選択します。

6. 次の値をコピーして安全に保管します。

   • キー: client_id として使用されます
   • Secret: client_secret として使用されます。
   注: キーとシークレットは画面に 1 回だけ表示されます。シークレットは後で再度表示できないため、ページを離れる前に両方の値をコピーしてください。キーとシークレットは、有効化から 1 年後に期限切れになります。

権限を確認する

API キー管理サービスは複数の 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 オブジェクトが返されます。

注: Reporting API は、アクセス トークンを Authorization: Bearer ヘッダーとしてのみ受け入れます。

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

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

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

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

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

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

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

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

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

Pub/Sub トピックの作成

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

1. GCP Console で、[Pub/Sub>トピック] に移動します。
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. [サービスを作成] をクリックします。
3. [関数] を選択します（インライン エディタを使用して関数を作成します）。

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

   設定	値
   サービス名	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）を確認します。
   注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。

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 のベース URL
   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	1 回の実行でバックフィルされる日数の上限
   SEEN_RETENTION_DAYS	7	状態に保持される重複除去キーの日数

   注: STATE_KEY は、ログ接頭辞の外を指すように意図的に設定されています。フィードはバケット URI のすべてのオブジェクトを取り込み、削除オプションが選択されると、転送したものを削除します。状態ファイルをその接頭辞から除外すると、ログデータとして取り込まれたり、削除されたりすることがなくなります。削除されると、コレクションがリセットされ、重複するイベントが再取り込みされます。

10. [変数とシークレット] セクションで、[リクエスト] に移動します。

    • リクエストのタイムアウト: 600 秒（10 分）を入力します。

11. [設定] タブに移動します。

    • [リソース] セクションで次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: 1 を選択します。

12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数]: 「0」と入力します。
    • インスタンスの最大数: 100 と入力します（または、予想される負荷に基づいて調整します）。

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

14. サービスが作成されるまで待ちます（1 ～ 2 分）。

15. サービスが作成されると、インライン コードエディタが自動的に開きます。

関数のコードを追加する

1. [エントリ ポイント] フィールドに「main」と入力します。
2. インライン コードエディタで、次の 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")}')
  

  注: このスクリプトは例として提供されており、公式にはサポートされていません。スクリプトの改善について、ぜひフィードバックをお寄せください。フィードバックを送信するには、このページで [Send Feedback > Google SecOps Security Product Feedback] をクリックします。

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

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

1. [デプロイ] をクリックして、関数を保存してデプロイします。

2. デプロイが完了するまで待ちます（2 ～ 3 分）。

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

Cloud Scheduler ジョブを作成する

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

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

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

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

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

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

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

頻度	CRON 式	ユースケース
毎日午前 2 時	0 2 * * *	標準（推奨）
毎日午前 6 時	0 6 * * *	レポートの決済が遅れるテナント
1 日 2 回	0 2,14 * * *	初回実行が失敗した場合に、同日の再試行を追加

実行時に、その日のデータがすでに記録されていることが判明した場合は、何も書き込まずに終了するため、追加の実行には 1 回の 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 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Proofpoint SER Logs）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [ProofPoint Secure Email Relay] を選択します。
7. [サービス アカウントを取得する] をクリックします。

8. 一意のサービス アカウントのメールアドレスが表示されます（例:）。

   [e-mail]
   

9. このメールアドレスをコピーして、次のステップで使用します。

   注: 各 Google SecOps インスタンスには一意のサービス アカウントがあります。他のドキュメントや例のサービス アカウントは使用しないでください。

10. [次へ] をクリックします。

11. 次の入力パラメータの値を指定します。

    • ストレージ バケットの URL: 接頭辞パスを含む Cloud Storage バケット URI を入力します。

      gs://proofpoint-ser-logs/ser-logs/
      

      • 次のように置き換えます。
        • proofpoint-ser-logs: Cloud Storage バケット名。
        • ser-logs: ログが保存されるオプションの接頭辞またはフォルダパス（ルートの場合は空のままにします）。
    注: URI の末尾には常にスラッシュ（/）を付けてください。

    • Source deletion option: 必要に応じて削除オプションを選択します。

      • ファイルを削除しない: ソースからファイルを削除しません（テストに推奨）。
      • 転送されたファイルと空のディレクトリを削除する: フェッチが正常に完了した後、ソースからファイルと空のディレクトリを削除します。

    • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます（デフォルトは 180 日）。

    • アセットの名前空間: アセットの名前空間

    • Ingestion labels: このフィードのイベントに適用されるラベル

12. [次へ] をクリックします。

13. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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

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

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

6. [保存] をクリックします。

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