Collecter les journaux Proofpoint Secure Email Relay

Compatible avec :

Ce document explique comment ingérer les journaux Proofpoint Secure Email Relay dans Google Security Operations à l'aide de Cloud Storage V2.

Proofpoint Secure Email Relay (SER) est un relais géré pour les e-mails générés par les applications, tels que les messages transactionnels et de notification. SER génère des rapports sur ce trafic via son API Reporting, l'une des API de configuration SER. L'API renvoie des chiffres d'utilisation agrégés : position du débit sous licence, volume et nombre de messages quotidiens, et compteurs pour chaque utilisateur du relais. L'analyseur mappe ces chiffres au modèle UDM (Unified Data Model), en les enregistrant sous forme de libellés à côté de l'identité de l'utilisateur du relais.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps
  • Un projet Google Cloud avec l'API Cloud Storage activée
  • Autorisations permettant de créer et de gérer des buckets Cloud Storage
  • Autorisations permettant de gérer les stratégies Identity and Access Management (IAM) sur les buckets Cloud Storage
  • Autorisations permettant de créer des services Cloud Run, des sujets Pub/Sub et des tâches Cloud Scheduler
  • Accès administrateur au portail Proofpoint Secure Email Relay, y compris l'autorisation de créer des clés API

Créer un bucket Cloud Storage

  1. Accédez à la consoleGoogle Cloud .
  2. Sélectionnez votre projet ou créez-en un.
  3. Dans le menu de navigation, accédez à Cloud Storage > Buckets.
  4. Cliquez sur Créer un bucket.

  5. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nommer votre bucket Saisissez un nom unique (par exemple, proofpoint-ser-logs).
    Type d'emplacement Choisissez en fonction de vos besoins (région, birégion ou multirégion).
    Emplacement Sélectionnez l'emplacement (par exemple, us-central1).
    Classe de stockage Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
    Access control (Contrôle des accès) Uniforme (recommandé)
    Outils de protection Facultatif : Activez la gestion des versions des objets ou une règle de conservation.

  6. Cliquez sur Créer.

Collecter les identifiants de l'API Proofpoint SER

L'accès à l'API de reporting est accordé par une clé API émise par le service Proofpoint API Key Management. La clé n'est pas créée dans la console SER elle-même. C'est pourquoi elle n'apparaît dans aucune des entrées de navigation SER.

Créer une clé API

  1. Connectez-vous au portail Proofpoint Secure Email Relay avec des identifiants d'administrateur.
  2. Ouvrez le sélecteur d'applications en haut à gauche.
  3. Accédez à Services > Gestion des clés API.
  4. Cliquez sur Create Key (Créer une clé).
  5. Sélectionnez Secure Email Relay comme produit.

  6. Copiez et stockez en lieu sûr les valeurs suivantes :

    • Clé : utilisée comme client_id
    • Secret : utilisé comme client_secret

Vérifier les autorisations

Le service Gestion des clés API est partagé entre plusieurs produits Proofpoint. C'est pourquoi vous y accédez via le sélecteur d'applications plutôt que via le menu de navigation SER. Pour accéder aux API de configuration SER, y compris l'API Reporting, vous devez sélectionner Secure Email Relay comme produit lorsque vous créez votre clé.

Si Gestion des clés API n'apparaît pas dans le sélecteur d'applications, cela signifie que votre compte ne dispose pas de l'autorisation de gérer les API. Contactez votre administrateur Proofpoint ou votre équipe de compte pour qu'ils vous accordent l'autorisation.

Tester l'accès à l'API

  • L'API Reporting utilise l'attribution des identifiants client OAuth 2.0. Échangez la clé et le secret contre un jeton du porteur, puis appelez l'API avec ce jeton.

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

Un appel réussi renvoie un objet JSON dont le membre data contient throughputLimit, licenseStartDate, licenseEndDate et les moyennes de débit.

Créer un compte de service pour la fonction Cloud Run

La fonction Cloud Run a besoin d'un compte de service autorisé à écrire dans un bucket Cloud Storage et à être appelée par Pub/Sub.

Créer le compte de service

  1. Dans la console GCP, accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Nom du compte de service : saisissez proofpoint-ser-collector-sa.
    • Description du compte de service : saisissez Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs.
  4. Cliquez sur Créer et continuer.
  5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
    1. Cliquez sur Sélectionner un rôle.
    2. Recherchez et sélectionnez Administrateur des objets de l'espace de stockage.
    3. Cliquez sur + Ajouter un autre rôle.
    4. Recherchez et sélectionnez Demandeur Cloud Run.
    5. Cliquez sur + Ajouter un autre rôle.
    6. Recherchez et sélectionnez Demandeur Cloud Functions.
  6. Cliquez sur Continuer.
  7. Cliquez sur OK.

Ces rôles sont requis pour :

  • Administrateur des objets Storage : écrire des journaux dans un bucket Cloud Storage et gérer les fichiers d'état
  • Demandeur Cloud Run : autorise Pub/Sub à appeler la fonction
  • Demandeur Cloud Functions : autorise l'appel de fonctions

Accorder des autorisations IAM sur le bucket Cloud Storage

Accordez au compte de service des autorisations en écriture sur le bucket Cloud Storage :

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (par exemple, proofpoint-ser-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (par exemple, proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Attribuer des rôles : sélectionnez Administrateur des objets Storage.
  6. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

Créez un sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.

  1. Dans la console GCP, accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. Fournissez les informations de configuration suivantes :
    • ID du sujet : saisissez proofpoint-ser-trigger.
    • Conservez les valeurs par défaut des autres paramètres.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

La fonction Cloud Run sera déclenchée par les messages Pub/Sub de Cloud Scheduler pour extraire les rapports d'utilisation de l'API SER Reporting et les écrire dans Cloud Storage.

  1. Dans la console GCP, accédez à Cloud Run.
  2. Cliquez sur Créer un service.
  3. Sélectionnez Fonction (utilisez un éditeur intégré pour créer une fonction).

  4. Dans la section Configurer, fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom du service proofpoint-ser-collector
    Région Sélectionnez la région correspondant à votre bucket Cloud Storage (par exemple, us-central1).
    Durée d'exécution Sélectionnez Python 3.12 ou version ultérieure.

  5. Dans la section Déclencheur (facultatif) :

    1. Cliquez sur + Ajouter un déclencheur.
    2. Sélectionnez Cloud Pub/Sub.
    3. Dans Sélectionner un sujet Cloud Pub/Sub, choisissez le sujet Pub/Sub (proofpoint-ser-trigger).
    4. Cliquez sur Enregistrer.

  6. Dans la section Authentification :

    1. Sélectionnez Exiger l'authentification.
    2. Consultez Identity and Access Management (IAM).

  7. Accédez à Conteneurs, mise en réseau, sécurité et développez cette section.

  8. Accédez à l'onglet Sécurité :

    • Compte de service : sélectionnez le compte de service (proofpoint-ser-collector-sa).

  9. Accédez à l'onglet Conteneurs :

    1. Cliquez sur Variables et secrets.
    2. Cliquez sur + Ajouter une variable pour chaque variable d'environnement :
    Nom de la variable Exemple de valeur Description
    GCS_BUCKET proofpoint-ser-logs Nom du bucket Cloud Storage
    GCS_PREFIX ser-logs Préfixe des fichiers journaux
    STATE_KEY ser-logs-state.json Chemin d'état, en dehors du préfixe du journal
    TOKEN_URL https://auth.proofpoint.com/v1/token Point de terminaison du jeton OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com URL de base de l'API SER Reporting
    CLIENT_ID your-key La clé issue de la gestion des clés API
    CLIENT_SECRET your-secret Le secret de la gestion des clés API
    PAGE_SIZE 50 Nombre d'utilisateurs Relay demandés par page
    MAX_PAGES 200 Plafond de pages pour la pagination des utilisateurs de relais
    REPORT_DAY_OFFSET 1 Rapport sur le jour correspondant à ce nombre de jours
    MAX_BACKFILL_DAYS 7 La plupart du temps, une seule exécution remplira les données manquantes.
    SEEN_RETENTION_DAYS 7 Nombre de jours pendant lesquels les clés de déduplication sont conservées dans l'état

  10. Dans la section Variables et secrets, accédez à Requêtes :

    • Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).

  11. Accédez à l'onglet Paramètres :

    • Dans la section Ressources :
      • Mémoire : sélectionnez 512 Mio ou plus.
      • CPU : sélectionnez 1.

  12. Dans la section Scaling de révision :

    • Nombre minimal d'instances : saisissez 0.
    • Nombre maximal d'instances : saisissez 100 (ou ajustez en fonction de la charge attendue).

  13. Cliquez sur Créer.

  14. Attendez que le service soit créé (1 à 2 minutes).

  15. Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter le code de la fonction

  1. Saisissez main dans le champ Point d'entrée.
  2. Dans l'éditeur de code intégré, créez deux fichiers :

  • Premier fichier : 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")}')

  • Deuxième fichier : requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Cliquez sur Déployer pour enregistrer et déployer la fonction.

  2. Attendez la fin du déploiement (deux à trois minutes).

Créer une tâche Cloud Scheduler

Cloud Scheduler publie des messages sur le sujet Pub/Sub à intervalles réguliers, ce qui déclenche la fonction Cloud Run.

  1. Dans la console GCP, accédez à Cloud Scheduler.
  2. Cliquez sur Créer une tâche.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom proofpoint-ser-collector-daily
    Région Sélectionnez la même région que la fonction Cloud Run.
    Fréquence 0 2 * * * (tous les jours, à 02:00)
    Fuseau horaire Sélectionnez un fuseau horaire (UTC recommandé).
    Type de cible Pub/Sub
    Sujet Sélectionnez le sujet Pub/Sub (proofpoint-ser-trigger).
    Corps du message {} (objet JSON vide)

  4. Cliquez sur Créer.

Options de fréquence de programmation

L'API Reporting agrège les données par jour calendaire. Par conséquent, un jour complet n'est collecté qu'une seule fois. Exécutez le job quotidiennement, une fois que le jour de reporting est terminé dans le fuseau horaire de votre locataire :

Fréquence Expression Cron Cas d'utilisation
Tous les jours à 2h 0 2 * * * Standard (recommandé)
Tous les jours à 06:00 0 6 * * * Locataires dont les rapports sont finalisés plus tard
Deux fois par jour 0 2,14 * * * Ajoute une nouvelle tentative d'exécution le même jour si la première a échoué

Une exécution qui trouve son jour déjà enregistré se termine sans rien écrire. Une exécution supplémentaire coûte donc un appel d'API et ne duplique jamais les données.

Tester l'intégration

  1. Dans la console Cloud Scheduler, recherchez votre job.
  2. Cliquez sur Exécuter de force pour déclencher le job manuellement.
  3. Patientez quelques secondes.
  4. Accédez à Cloud Run > Services.
  5. Cliquez sur le nom de votre fonction (proofpoint-ser-collector).
  6. Cliquez sur l'onglet Journaux.

  7. Vérifiez que la fonction s'est exécutée correctement. Par exemple :

    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. Accédez à Cloud Storage > Buckets.

  9. Cliquez sur le nom de votre bucket (proofpoint-ser-logs).

  10. Accédez au dossier de préfixe (ser-logs/).

  11. Vérifiez qu'un fichier .ndjson a été créé avec le code temporel actuel.

Si vous constatez des erreurs dans les journaux :

  • HTTP 400 du point de terminaison de jeton : la clé ou le code secret sont incorrects, ou la clé a expiré. invalid_client signifie que la paire a été refusée. Créez une clé de remplacement.
  • HTTP 401 de l'API Reporting : le jeton du porteur n'a pas été envoyé ou n'est plus valide.
  • HTTP 429 : limitation du débit. La fonction effectue des tentatives avec un intervalle exponentiel et fait échouer l'exécution si la limite persiste, laissant le jour à la prochaine exécution.
  • Variables d'environnement manquantes : vérifiez que toutes les variables requises sont définies.

Récupérer le compte de service Google SecOps

Google SecOps utilise un compte de service unique pour lire les données de votre bucket Cloud Storage. Vous devez accorder à ce compte de service l'accès à votre bucket.

Obtenir l'adresse e-mail du compte de service

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Add New Feed (Ajouter un flux).
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Proofpoint SER Logs).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez ProofPoint Secure Email Relay comme Type de journal.
  7. Cliquez sur Obtenir un compte de service.

  8. Une adresse e-mail unique pour le compte de service s'affiche, par exemple :

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

  9. Copiez cette adresse e-mail pour l'utiliser à l'étape suivante.

  10. Cliquez sur Suivant.

  11. Spécifiez les valeurs des paramètres d'entrée suivants :

    • URL du bucket de stockage : saisissez l'URI du bucket Cloud Storage avec le préfixe du chemin d'accès :

      gs://proofpoint-ser-logs/ser-logs/
      • Remplacez :
        • proofpoint-ser-logs : nom de votre bucket Cloud Storage.
        • ser-logs : préfixe ou chemin d'accès facultatif au dossier dans lequel les journaux sont stockés (laisser vide pour la racine).

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Ne jamais supprimer les fichiers : ne supprime jamais les fichiers de la source (recommandé pour les tests).
      • Supprimer les fichiers transférés et les répertoires vides : supprimez les fichiers et les répertoires vides de la source une fois l'extraction réussie.

    • Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours (180 jours par défaut).

    • Espace de noms de l'élément : espace de noms de l'élément

    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux

  12. Cliquez sur Suivant.

  13. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps a besoin de deux rôles sur votre bucket Cloud Storage : Lecteur des objets Storage pour lire les objets journaux et un rôle au niveau du bucket pour lire les métadonnées du bucket.

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (par exemple, proofpoint-ser-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
    • Attribuer des rôles : sélectionnez les deux options suivantes :
      • Lecteur des objets Storage : lit les objets journaux.
      • Lecteur des anciens buckets Storage : lit les métadonnées du bucket. Si vous avez sélectionné l'option de suppression Supprimer les fichiers transférés et les répertoires vides, sélectionnez plutôt Storage Legacy Bucket Writer, qui accorde également l'autorisation de suppression.

  6. Cliquez sur Enregistrer.

Table de mappage UDM

Champ de journal Mappage UDM Logique
acceptedMessages_label additional.fields Fusionné
acceptedThroughput_label additional.fields Fusionné
average30DayThroughput_label additional.fields Fusionné
average7DayThroughput_label additional.fields Fusionné
averageDailyThroughput_label additional.fields Fusionné
avgAcceptedMessageSize_label additional.fields Fusionné
blockedMessages_label additional.fields Fusionné
data_totalThroughput_label additional.fields Fusionné
deliveredMessages_label additional.fields Fusionné
licenseEndDate_label additional.fields Fusionné
licenseStartDate_label additional.fields Fusionné
map_label additional.fields Fusionné
metadata_totalThroughput_label additional.fields Fusionné
quarantinedMessages_label additional.fields Fusionné
rejectedMessages_label additional.fields Fusionné
remainingThroughput_label additional.fields Fusionné
requestedMessages_label additional.fields Fusionné
requestedThroughput_label additional.fields Fusionné
sentMessages_label additional.fields Fusionné
throughputForecast_label additional.fields Fusionné
throughputLimit_label additional.fields Fusionné
throughput_label additional.fields Fusionné
totalMessages_label additional.fields Fusionné
undeliveredMessages_label additional.fields Fusionné
data.name metadata.description Mappé directement
desc metadata.description Mappé directement
data.date metadata.event_timestamp Analysé comme yyyy-MM-dd
event_type metadata.event_type Mappé directement
fromEnvelope network.email.bounce_address Mappé directement
fromHeader network.email.from Mappé directement
applicationName principal.administrative_domain Mappé directement
principal_host principal.asset.hostname Mappé directement
principal_host principal.hostname Mappé directement
principal_port principal.port Mappé directement
data.relayUserId principal.user.product_object_id Mappé directement
userId principal.user.product_object_id Mappé directement
applicationUserName principal.user.user_display_name Mappé directement
senderName target.administrative_domain Mappé directement
senderId target.user.product_object_id Mappé directement
N/A metadata.product_name Constante : PROOFPOINT SER
N/A metadata.vendor_name Constante : PROOFPOINT

Journal des modifications

Afficher le journal des modifications pour ce parseur

Vous avez encore besoin d'aide ? Obtenez des réponses auprès des membres de la communauté et des professionnels Google SecOps.