Proofpoint Secure Email Relay-Logs erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Proofpoint Secure Email Relay-Logs mit Cloud Storage V2 in Google Security Operations aufnehmen.

Proofpoint Secure Email Relay (SER) ist ein verwaltetes Relais für von Anwendungen generierte E-Mails wie Transaktions- und Benachrichtigungsnachrichten. SER meldet diesen Traffic über die Reporting API, eine der SER-Konfigurations-APIs. Die API gibt aggregierte Nutzungszahlen zurück: die Position des lizenzierten Durchsatzes, das tägliche Volumen und die Anzahl der Nachrichten sowie Zähler für jeden Relay-Nutzer. Der Parser ordnet diese Zahlen dem einheitlichen Datenmodell (Unified Data Model, UDM) zu und erfasst sie als Labels neben der Identität des Relay-Nutzers.

Hinweis

Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Ein Google Cloud -Projekt mit aktivierter Cloud Storage API
  • Berechtigungen zum Erstellen und Verwalten von Cloud Storage-Buckets
  • Berechtigungen zum Verwalten von IAM-Richtlinien (Identity and Access Management) für Cloud Storage-Buckets
  • Berechtigungen zum Erstellen von Cloud Run-Diensten, Pub/Sub-Themen und Cloud Scheduler-Jobs
  • Administratorzugriff auf das Proofpoint Secure Email Relay-Portal, einschließlich der Berechtigung zum Erstellen von API-Schlüsseln

Cloud Storage-Bucket erstellen

  1. Rufen Sie die Google Cloud Console auf.
  2. Wählen Sie Ihr Projekt aus oder erstellen Sie ein neues.
  3. Rufen Sie im Navigationsmenü Cloud Storage > Buckets auf.
  4. Klicken Sie auf Bucket erstellen.

  5. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Bucket benennen Geben Sie einen global eindeutigen Namen ein, z. B. proofpoint-ser-logs.
    Standorttyp Wählen Sie die Option aus, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region).
    Standort Wählen Sie den Standort aus, z. B. us-central1.
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffskontrolle Einheitlich (empfohlen)
    Schutzmaßnahmen Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

Proofpoint SER API-Anmeldedaten erfassen

Der Zugriff auf die Reporting API wird über einen API-Schlüssel gewährt, der vom Proofpoint API Key Management Service ausgestellt wird. Der Schlüssel wird nicht in der SER-Konsole selbst erstellt. Daher wird er unter keinem der SER-Navigationseinträge angezeigt.

API-Schlüssel erstellen

  1. Melden Sie sich mit Administratoranmeldedaten im Proofpoint Secure Email Relay-Portal an.
  2. Öffnen Sie den App-Schnellzugriff oben links.
  3. Rufen Sie Dienste > API-Schlüsselverwaltung auf.
  4. Klicken Sie auf Schlüssel erstellen.
  5. Wählen Sie Secure Email Relay als Produkt aus.

  6. Kopieren Sie die folgenden Werte und speichern Sie sie sicher:

    • Schlüssel: Wird als client_id verwendet.
    • Secret: Wird als client_secret verwendet.

Berechtigungen prüfen

Der Dienst API Key Management wird von mehreren Proofpoint-Produkten gemeinsam genutzt. Daher greifen Sie über den App-Schnellzugriff und nicht über das SER-Navigationsmenü darauf zu. Wenn Sie Zugriff auf die SER-Konfigurations-APIs, einschließlich der Reporting API, erhalten möchten, müssen Sie beim Erstellen Ihres Schlüssels Secure Email Relay als Produkt auswählen.

Wenn API-Schlüsselverwaltung nicht im App-Schnellzugriff angezeigt wird, fehlt Ihrem Konto die Berechtigung zur API-Verwaltung. Wenden Sie sich an Ihren Proofpoint-Administrator oder Ihr Kontoteam, damit diese Ihnen die Berechtigung gewähren können.

API-Zugriff testen

  • Die Reporting API verwendet den OAuth 2.0-Grant-Typ „Clientanmeldedaten“. Tauschen Sie den Schlüssel und das Secret gegen ein Bearertoken ein und rufen Sie dann die API mit diesem Token auf.

    # 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"

Bei einem erfolgreichen Aufruf wird ein JSON-Objekt zurückgegeben, dessen data-Element throughputLimit, licenseStartDate, licenseEndDate und die durchschnittlichen Durchsatzwerte enthält.

Dienstkonto für die Cloud Run-Funktion erstellen

Die Cloud Run-Funktion benötigt ein Dienstkonto mit Berechtigungen zum Schreiben in einen Cloud Storage-Bucket und zum Aufrufen durch Pub/Sub.

Erstellen Sie das Dienstkonto:

  1. Wechseln Sie in der GCP Console zu IAM & Verwaltung > Dienstkonten.
  2. Klicken Sie auf Dienstkonto erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Name des Dienstkontos: Geben Sie proofpoint-ser-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs ein.
  4. Klicken Sie auf Erstellen und fortfahren.
  5. Fügen Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen die folgenden Rollen hinzu:
    1. Klicken Sie auf Rolle auswählen.
    2. Suchen Sie nach Storage-Objekt-Administrator und wählen Sie die Rolle aus.
    3. Klicken Sie auf + Weitere Rolle hinzufügen.
    4. Suchen Sie nach Cloud Run Invoker und wählen Sie die Rolle aus.
    5. Klicken Sie auf + Weitere Rolle hinzufügen.
    6. Suchen Sie nach Cloud Functions Invoker und wählen Sie die Rolle aus.
  6. Klicken Sie auf Weiter.
  7. Klicken Sie auf Fertig.

Diese Rollen sind erforderlich für:

  • Storage-Objekt-Administrator: Protokolle in einen Cloud Storage-Bucket schreiben und Statusdateien verwalten
  • Cloud Run-Aufrufer: Pub/Sub darf die Funktion aufrufen.
  • Cloud Functions-Invoker: Funktionsaufruf zulassen

IAM-Berechtigungen für den Cloud Storage-Bucket gewähren

Gewähren Sie dem Dienstkonto Schreibberechtigungen für den Cloud Storage-Bucket:

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets, z. B. proofpoint-ser-logs.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos ein (z. B. proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Rollen zuweisen: Wählen Sie Storage-Objekt-Administrator aus.
  6. Klicken Sie auf Speichern.

Pub/Sub-Thema erstellen

Erstellen Sie ein Pub/Sub-Thema, in dem Cloud Scheduler veröffentlicht und das von der Cloud Run-Funktion abonniert wird.

  1. Rufen Sie in der GCP Console Pub/Sub > Themen auf.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Themen-ID: Geben Sie proofpoint-ser-trigger ein.
    • Übernehmen Sie die anderen Einstellungen.
  4. Klicken Sie auf Erstellen.

Cloud Run-Funktion zum Erfassen von Logs erstellen

Die Cloud Run-Funktion wird durch Pub/Sub-Nachrichten von Cloud Scheduler ausgelöst, um Nutzungsberichte von der SER Reporting API abzurufen und in Cloud Storage zu schreiben.

  1. Rufen Sie in der GCP Console Cloud Run auf.
  2. Klicken Sie auf Dienst erstellen.
  3. Wählen Sie Funktion aus, um eine Funktion mit einem Inline-Editor zu erstellen.

  4. Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Dienstname proofpoint-ser-collector
    Region Wählen Sie die Region aus, die Ihrem Cloud Storage-Bucket entspricht (z. B. us-central1).
    Laufzeit Wählen Sie Python 3.12 oder höher aus.

  5. Im Abschnitt Trigger (optional):

    1. Klicken Sie auf + Trigger hinzufügen.
    2. Wählen Sie Cloud Pub/Sub aus.
    3. Wählen Sie unter Cloud Pub/Sub-Thema auswählen das Pub/Sub-Thema (proofpoint-ser-trigger) aus.
    4. Klicken Sie auf Speichern.

  6. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Identitäts- und Zugriffsverwaltung

  7. Rufen Sie Container, Netzwerk, Sicherheit auf und maximieren Sie den Bereich.

  8. Rufen Sie den Tab Sicherheit auf:

    • Dienstkonto: Wählen Sie das Dienstkonto aus (proofpoint-ser-collector-sa).

  9. Rufen Sie den Tab Container auf:

    1. Klicken Sie auf Variablen und Secrets.
    2. Klicken Sie für jede Umgebungsvariable auf + Variable hinzufügen:
    Variablenname Beispielwert Beschreibung
    GCS_BUCKET proofpoint-ser-logs Name des Cloud Storage-Buckets
    GCS_PREFIX ser-logs Präfix für Protokolldateien
    STATE_KEY ser-logs-state.json Statuspfad außerhalb des Log-Präfixes
    TOKEN_URL https://auth.proofpoint.com/v1/token OAuth 2.0-Tokenendpunkt
    API_BASE https://reporting.ser.proofpoint.com SER Reporting API-Basis-URL
    CLIENT_ID your-key Der Schlüssel aus der API-Schlüsselverwaltung
    CLIENT_SECRET your-secret Das Secret aus der API-Schlüsselverwaltung
    PAGE_SIZE 50 Angeforderte Relay-Nutzer pro Seite
    MAX_PAGES 200 Seitenlimit für die Paginierung von Relay-Nutzern
    REPORT_DAY_OFFSET 1 Bericht für den Tag vor dieser Anzahl von Tagen erstellen
    MAX_BACKFILL_DAYS 7 An den meisten Tagen wird nur ein Lauf nachgetragen.
    SEEN_RETENTION_DAYS 7 Anzahl der Tage, die Deduplizierungsschlüssel im Status beibehalten werden

  10. Gehen Sie im Bereich Variablen und Secrets zu Anfragen:

    • Zeitüberschreitung bei Anfrage: Geben Sie 600 Sekunden (10 Minuten) ein.

  11. Rufen Sie den Tab Einstellungen auf:

    • Im Abschnitt Ressourcen:
      • Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
      • CPU: Wählen Sie 1 aus.

  12. Im Abschnitt Versionsskalierung:

    • Mindestanzahl von Instanzen: Geben Sie 0 ein.
    • Maximale Anzahl von Instanzen: Geben Sie 100 ein (oder passen Sie den Wert an die erwartete Last an).

  13. Klicken Sie auf Erstellen.

  14. Warten Sie ein bis zwei Minuten, bis der Dienst erstellt wurde.

  15. Nachdem der Dienst erstellt wurde, wird automatisch der Inline-Code-Editor geöffnet.

Funktionscode hinzufügen

  1. Geben Sie main in das Feld Einstiegspunkt ein.
  2. Erstellen Sie im Inline-Codeeditor zwei Dateien:

  • Erste Datei: 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")}')

  • Zweite Datei – requirements.txt::

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Klicken Sie auf Bereitstellen, um die Funktion zu speichern und bereitzustellen.

  2. Warten Sie, bis die Bereitstellung abgeschlossen ist (2–3 Minuten).

Cloud Scheduler-Job erstellen

Cloud Scheduler veröffentlicht in regelmäßigen Abständen Nachrichten im Pub/Sub-Thema und löst so die Cloud Run-Funktion aus.

  1. Rufen Sie in der GCP Console Cloud Scheduler auf.
  2. Klicken Sie auf Job erstellen.

  3. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Name proofpoint-ser-collector-daily
    Region Dieselbe Region wie für die Cloud Run-Funktion auswählen
    Frequenz 0 2 * * * (täglich um 02:00 Uhr)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema Wählen Sie das Pub/Sub-Thema aus (proofpoint-ser-trigger).
    Inhalt der Nachricht {} (leeres JSON-Objekt)

  4. Klicken Sie auf Erstellen.

Optionen für die Häufigkeit des Zeitplans

In der Reporting API werden Daten nach Kalendertag zusammengefasst. Ein abgeschlossener Tag wird also nur einmal erfasst. Führen Sie den Job täglich aus, nachdem der Berichtstag in der Zeitzone Ihres Mandanten abgeschlossen ist:

Häufigkeit Cron-Ausdruck Anwendungsfall
Täglich um 02:00 Uhr 0 2 * * * Standard (empfohlen)
Täglich um 06:00 Uhr 0 6 * * * Mandanten, deren Berichte später abgerechnet werden
Zweimal täglich 0 2,14 * * * Fügt einen Wiederholungsversuch am selben Tag hinzu, wenn der erste Lauf fehlgeschlagen ist

Wenn ein Lauf feststellt, dass der Tag bereits aufgezeichnet wurde, wird er beendet, ohne etwas zu schreiben. Ein zusätzlicher Lauf kostet also einen API-Aufruf und dupliziert niemals Daten.

Integration testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job.
  2. Klicken Sie auf Force run (Ausführung erzwingen), um den Job manuell auszulösen.
  3. Warten Sie einige Sekunden.
  4. Rufen Sie Cloud Run > Dienste auf.
  5. Klicken Sie auf den Namen Ihrer Funktion (proofpoint-ser-collector).
  6. Klicken Sie auf den Tab Logs.

  7. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde. Darauf sollten Sie achten:

    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. Rufen Sie Cloud Storage > Buckets auf.

  9. Klicken Sie auf den Namen Ihres Buckets (proofpoint-ser-logs).

  10. Rufen Sie den Präfixordner (ser-logs/) auf.

  11. Prüfen Sie, ob eine neue .ndjson-Datei mit dem aktuellen Zeitstempel erstellt wurde.

Wenn Sie Fehler in den Logs sehen:

  • HTTP 400 vom Token-Endpunkt: Der Schlüssel oder das Secret ist falsch oder der Schlüssel ist abgelaufen. invalid_client bedeutet, dass das Paar abgelehnt wurde. Erstellen Sie einen Ersatzschlüssel.
  • HTTP 401 von der Reporting API: Das Bearer-Token wurde nicht gesendet oder ist nicht mehr gültig.
  • HTTP 429: Ratenbegrenzung. Die Funktion wiederholt den Vorgang mit exponentieller Rücksetzung und lässt den Lauf fehlschlagen, wenn das Limit weiterhin überschritten wird. Der Tag wird dann im nächsten Lauf verarbeitet.
  • Fehlende Umgebungsvariablen: Prüfen Sie, ob alle erforderlichen Variablen festgelegt sind.

Google SecOps-Dienstkonto abrufen

Google SecOps verwendet ein eindeutiges Dienstkonto, um Daten aus Ihrem Cloud Storage-Bucket zu lesen. Sie müssen diesem Dienstkonto Zugriff auf Ihren Bucket gewähren.

E‑Mail-Adresse des Dienstkontos abrufen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Feed hinzufügen.
  3. Klicken Sie auf Einzelnen Feed konfigurieren.
  4. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. Proofpoint SER Logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie ProofPoint Secure Email Relay als Logtyp aus.
  7. Klicken Sie auf Dienstkonto abrufen.

  8. Es wird eine eindeutige E-Mail-Adresse für das Dienstkonto angezeigt, z. B.:

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

  9. Kopieren Sie diese E-Mail-Adresse für den nächsten Schritt.

  10. Klicken Sie auf Weiter.

  11. Geben Sie Werte für die folgenden Eingabeparameter an:

    • Storage-Bucket-URL: Geben Sie den Cloud Storage-Bucket-URI mit dem Präfixpfad ein:

      gs://proofpoint-ser-logs/ser-logs/
      • Ersetzen Sie:
        • proofpoint-ser-logs: Name Ihres Cloud Storage-Bucket.
        • ser-logs: Optionales Präfix oder Ordnerpfad, in dem Logs gespeichert werden (leer lassen für Stammverzeichnis).

    • Option zum Löschen der Quelle: Wählen Sie die gewünschte Löschoption aus:

      • Dateien nie löschen: Dateien werden niemals aus der Quelle gelöscht (empfohlen für Tests).
      • Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach einem erfolgreichen Abruf aus der Quelle gelöscht.

    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden (Standardwert: 180 Tage)

    • Asset-Namespace: Der Asset-Namespace

    • Labels für Datenaufnahme: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll

  12. Klicken Sie auf Weiter.

  13. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt zwei Rollen für Ihren Cloud Storage-Bucket: Storage-Objekt-Betrachter zum Lesen der Logobjekte und eine Rolle auf Bucket-Ebene zum Lesen der Bucket-Metadaten.

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets, z. B. proofpoint-ser-logs.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Fügen Sie die E‑Mail-Adresse des Google SecOps-Dienstkontos ein.
    • Rollen zuweisen: Wählen Sie die beiden folgenden Optionen aus:
      • Storage-Objekt-Betrachter: Liest die Logobjekte.
      • Leser alter Storage-Buckets: Liest die Bucket-Metadaten. Wenn Sie die Löschoption Übertragene Dateien und leere Verzeichnisse löschen ausgewählt haben, wählen Sie stattdessen Storage Legacy Bucket Writer aus. Damit wird auch die Löschberechtigung erteilt.

  6. Klicken Sie auf Speichern.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
acceptedMessages_label additional.fields Zusammengeführt
acceptedThroughput_label additional.fields Zusammengeführt
average30DayThroughput_label additional.fields Zusammengeführt
average7DayThroughput_label additional.fields Zusammengeführt
averageDailyThroughput_label additional.fields Zusammengeführt
avgAcceptedMessageSize_label additional.fields Zusammengeführt
blockedMessages_label additional.fields Zusammengeführt
data_totalThroughput_label additional.fields Zusammengeführt
deliveredMessages_label additional.fields Zusammengeführt
licenseEndDate_label additional.fields Zusammengeführt
licenseStartDate_label additional.fields Zusammengeführt
map_label additional.fields Zusammengeführt
metadata_totalThroughput_label additional.fields Zusammengeführt
quarantinedMessages_label additional.fields Zusammengeführt
rejectedMessages_label additional.fields Zusammengeführt
remainingThroughput_label additional.fields Zusammengeführt
requestedMessages_label additional.fields Zusammengeführt
requestedThroughput_label additional.fields Zusammengeführt
sentMessages_label additional.fields Zusammengeführt
throughputForecast_label additional.fields Zusammengeführt
throughputLimit_label additional.fields Zusammengeführt
throughput_label additional.fields Zusammengeführt
totalMessages_label additional.fields Zusammengeführt
undeliveredMessages_label additional.fields Zusammengeführt
data.name metadata.description Direkt zugeordnet
desc metadata.description Direkt zugeordnet
data.date metadata.event_timestamp Geparst als yyyy-MM-dd
event_type metadata.event_type Direkt zugeordnet
fromEnvelope network.email.bounce_address Direkt zugeordnet
fromHeader network.email.from Direkt zugeordnet
applicationName principal.administrative_domain Direkt zugeordnet
principal_host principal.asset.hostname Direkt zugeordnet
principal_host principal.hostname Direkt zugeordnet
principal_port principal.port Direkt zugeordnet
data.relayUserId principal.user.product_object_id Direkt zugeordnet
userId principal.user.product_object_id Direkt zugeordnet
applicationUserName principal.user.user_display_name Direkt zugeordnet
senderName target.administrative_domain Direkt zugeordnet
senderId target.user.product_object_id Direkt zugeordnet
– metadata.product_name Konstante: PROOFPOINT SER
– metadata.vendor_name Konstante: PROOFPOINT

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten