Raccogliere i log di Proofpoint Secure Email Relay

Supportato in:

Questo documento spiega come importare i log di Proofpoint Secure Email Relay in Google Security Operations utilizzando Cloud Storage V2.

Proofpoint Secure Email Relay (SER) è un relè gestito per la posta generata dalle applicazioni, ad esempio messaggi transazionali e di notifica. SER genera report su questo traffico tramite la relativa API Reporting, una delle API di configurazione SER. L'API restituisce cifre di utilizzo aggregate: la posizione del throughput con licenza, il volume giornaliero e i conteggi dei messaggi, nonché i contatori per ogni utente di inoltro. Il parser mappa queste cifre nel modello Unified Data Model (UDM), registrandole come etichette insieme all'identità dell'utente di inoltro.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto Google Cloud con l'API Storage di Cloud abilitata
  • Autorizzazioni per creare e gestire bucket Cloud Storage
  • Autorizzazioni per gestire i criteri IAM (Identity and Access Management) nei bucket Cloud Storage
  • Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso amministratore al portale Proofpoint Secure Email Relay, inclusa l'autorizzazione per creare chiavi API

Crea un bucket Cloud Storage

  1. Vai alla consoleGoogle Cloud .
  2. Seleziona il tuo progetto o creane uno nuovo.
  3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
  4. Fai clic su Crea bucket.

  5. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio proofpoint-ser-logs).
    Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
    Località Seleziona la posizione (ad esempio, us-central1)
    Classe di archiviazione Standard (consigliato per i log a cui si accede di frequente)
    Controllo degli accessi Uniforme (consigliato)
    Strumenti di protezione (Facoltativo) Attiva il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Raccogli le credenziali API Proofpoint SER

L'accesso all'API Reporting viene concesso da una chiave API emessa dal servizio Key Management Service di Proofpoint. La chiave non viene creata all'interno della console SER, motivo per cui non viene visualizzata in nessuna delle voci di navigazione SER.

crea una chiave API

  1. Accedi al portale Proofpoint Secure Email Relay con le credenziali di amministratore.
  2. Apri il selettore di app nell'angolo in alto a sinistra.
  3. Vai a Servizi > Gestione delle chiavi API.
  4. Fai clic su Crea chiave.
  5. Seleziona Secure Email Relay come prodotto.

  6. Copia e memorizza in modo sicuro i seguenti valori:

    • Chiave: utilizzata come client_id
    • Secret: utilizzato come client_secret

Verifica le autorizzazioni

Il servizio Gestione chiavi API è condiviso tra più prodotti Proofpoint, motivo per cui vi accedi tramite il selettore di app anziché tramite il menu di navigazione SER. Per accedere alle API di configurazione SER, inclusa l'API Reporting, devi selezionare Secure Email Relay come prodotto durante la creazione della chiave.

Se Gestione chiavi API non viene visualizzata nel selettore di app, il tuo account non dispone dell'autorizzazione di gestione delle API. Contatta l'amministratore o il team dedicato all'account Proofpoint per ottenere l'autorizzazione.

Testare l'accesso API

  • L'API Reporting utilizza la concessione delle credenziali client OAuth 2.0. Scambia la chiave e il secret con un token di autenticazione, quindi chiama l'API con quel token.

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

Una chiamata riuscita restituisce un oggetto JSON il cui membro data contiene throughputLimit, licenseStartDate, licenseEndDate e le medie di velocità effettiva.

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni per scrivere in un bucket Cloud Storage e per essere richiamato da Pub/Sub.

Crea il account di servizio

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service account.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome del service account: inserisci proofpoint-ser-collector-sa
    • Descrizione service account: inserisci Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
    1. Fai clic su Seleziona un ruolo.
    2. Cerca e seleziona Storage Object Admin.
    3. Fai clic su + Aggiungi un altro ruolo.
    4. Cerca e seleziona Cloud Run Invoker.
    5. Fai clic su + Aggiungi un altro ruolo.
    6. Cerca e seleziona Invoker di Cloud Functions.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono necessari per:

  • Storage Object Admin: scrive i log in un bucket Cloud Storage e gestisce i file di stato
  • Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
  • Cloud Functions Invoker: consente la chiamata di funzioni

Concedere le autorizzazioni IAM sul bucket Cloud Storage

Concedi all'account di servizio le autorizzazioni di scrittura sul bucket Cloud Storage:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (ad esempio proofpoint-ser-logs).
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'email del account di servizio (ad esempio, proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea un argomento Pub/Sub

Crea un argomento Pub/Sub a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

  1. Nella console GCP, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci proofpoint-ser-trigger
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run verrà attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i report sull'utilizzo dall'API SER Reporting e scriverli in Cloud Storage.

  1. Nella console GCP, vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione (usa un editor in linea per creare una funzione).

  4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio proofpoint-ser-collector
    Regione Seleziona la regione corrispondente al bucket Cloud Storage (ad esempio, us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive

  5. Nella sezione Trigger (facoltativo):

    1. Fai clic su + Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento Pub/Sub (proofpoint-ser-trigger).
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Controlla Identity and Access Management (IAM).

  7. Vai a Container, networking, sicurezza ed espandilo.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio (proofpoint-ser-collector-sa).

  9. Vai alla scheda Container:

    1. Fai clic su Variabili e secret.
    2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
    Nome variabile Valore di esempio Descrizione
    GCS_BUCKET proofpoint-ser-logs Nome del bucket Cloud Storage
    GCS_PREFIX ser-logs Prefisso per i file di log
    STATE_KEY ser-logs-state.json Percorso dello stato, al di fuori del prefisso del log
    TOKEN_URL https://auth.proofpoint.com/v1/token Endpoint del token OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com URL di base dell'API SER Reporting
    CLIENT_ID your-key La chiave della gestione delle chiavi API
    CLIENT_SECRET your-secret Il secret di Gestione chiavi API
    PAGE_SIZE 50 Utenti con disabilità uditive richiesti per pagina
    MAX_PAGES 200 Limite di pagine per la paginazione degli utenti di inoltro
    REPORT_DAY_OFFSET 1 Report sul giorno di questo numero di giorni fa
    MAX_BACKFILL_DAYS 7 La maggior parte dei giorni un'unica esecuzione eseguirà il backfill
    SEEN_RETENTION_DAYS 7 Giorni di chiavi di deduplicazione mantenute nello stato

  10. Nella sezione Variabili e secret, vai a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti)

  11. Vai alla scheda Impostazioni:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1

  12. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0
    • Numero massimo di istanze: inserisci 100 (o modifica in base al carico previsto)

  13. Fai clic su Crea.

  14. Attendi la creazione del servizio (1-2 minuti).

  15. Dopo aver creato il servizio, si aprirà automaticamente l'editor di codice incorporato.

Aggiungi il codice per la funzione

  1. Inserisci main nel campo Entry point (Punto di ingresso).
  2. Nell'editor di codice incorporato, crea due file:

  • Primo file - 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")}')

  • Secondo file:requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

  2. Attendi il completamento del deployment (2-3 minuti).

Crea un job Cloud Scheduler

Cloud Scheduler pubblicherà messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

  1. Nella console GCP, vai a Cloud Scheduler.
  2. Fai clic su Crea job.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome proofpoint-ser-collector-daily
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza 0 2 * * * (ogni giorno alle ore 02:00)
    Fuso orario Seleziona il fuso orario (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento Pub/Sub (proofpoint-ser-trigger).
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Opzioni di frequenza di pianificazione

L'API Reporting esegue l'aggregazione per giorno di calendario, quindi un giorno completato viene raccolto una sola volta. Esegui il job ogni giorno, dopo la chiusura del giorno del report nel fuso orario del tenant:

Frequenza Espressione cron Caso d'uso
Ogni giorno alle 02:00 0 2 * * * Standard (consigliato)
Ogni giorno alle 06:00 0 6 * * * Inquilini il cui report viene saldato in un secondo momento
Due volte al giorno 0 2,14 * * * Aggiunge un nuovo tentativo nello stesso giorno se la prima esecuzione non è riuscita

Un'esecuzione che rileva che il giorno è già stato registrato termina senza scrivere nulla, quindi un'esecuzione aggiuntiva costa una chiamata API e non duplica mai i dati.

Testare l'integrazione

  1. Nella console Cloud Scheduler, trova il tuo job.
  2. Fai clic su Forza esecuzione per attivare il job manualmente.
  3. Attendi qualche secondo.
  4. Vai a Cloud Run > Servizi.
  5. Fai clic sul nome della funzione (proofpoint-ser-collector).
  6. Fai clic sulla scheda Log.

  7. Verifica che la funzione sia stata eseguita correttamente. Cerca:

    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. Vai a Cloud Storage > Bucket.

  9. Fai clic sul nome del bucket (proofpoint-ser-logs).

  10. Vai alla cartella del prefisso (ser-logs/).

  11. Verifica che sia stato creato un nuovo file .ndjson con il timestamp corrente.

Se visualizzi errori nei log:

  • HTTP 400 dall'endpoint token: la chiave o il segreto non sono corretti oppure la chiave è scaduta. invalid_client significa che la coppia è stata rifiutata; crea una chiave sostitutiva.
  • HTTP 401 dall'API Reporting: il token di autenticazione non è stato inviato o non è più valido.
  • HTTP 429: limitazione della frequenza. La funzione esegue nuovi tentativi con backoff e l'esecuzione non va a buon fine se il limite persiste, lasciando il giorno all'esecuzione successiva.
  • Variabili di ambiente mancanti: controlla che tutte le variabili richieste siano impostate.

Recuperare il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal tuo bucket Cloud Storage. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'email del account di servizio

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Proofpoint SER Logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona ProofPoint Secure Email Relay come Tipo di log.
  7. Fai clic su Ottieni service account.

  8. Verrà visualizzata un'email del account di servizio univoca, ad esempio:

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

  9. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

  10. Fai clic su Avanti.

  11. Specifica i valori per i seguenti parametri di input:

    • URL del bucket di archiviazione: inserisci l'URI del bucket Cloud Storage con il percorso del prefisso:

      gs://proofpoint-ser-logs/ser-logs/
      • Sostituisci:
        • proofpoint-ser-logs: il nome del bucket Cloud Storage.
        • ser-logs: (Facoltativo) prefisso o percorso della cartella in cui sono archiviati i log (lascia vuoto per la radice).

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Non eliminare mai i file: non eliminare mai i file dall'origine (opzione consigliata per i test).
      • Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dall'origine al termine di un recupero riuscito.

    • Età massima file: includi i file modificati nell'ultimo numero di giorni (il valore predefinito è 180 giorni)

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed

  12. Fai clic su Avanti.

  13. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps ha bisogno di due ruoli nel tuo bucket Cloud Storage: Visualizzatore oggetti Storage per leggere gli oggetti log e un ruolo a livello di bucket per leggere i metadati del bucket.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (ad esempio proofpoint-ser-logs).
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'email del account di servizio Google SecOps
    • Assegna ruoli: seleziona entrambi i seguenti ruoli:
      • Visualizzatore oggetti Storage: legge gli oggetti log.
      • Storage Legacy Bucket Reader: legge i metadati del bucket. Se hai selezionato l'opzione di eliminazione Elimina file trasferiti e directory vuote, seleziona Storage Legacy Bucket Writer, che concede anche l'autorizzazione di eliminazione.

  6. Fai clic su Salva.

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
acceptedMessages_label additional.fields Unita
acceptedThroughput_label additional.fields Unita
average30DayThroughput_label additional.fields Unita
average7DayThroughput_label additional.fields Unita
averageDailyThroughput_label additional.fields Unita
avgAcceptedMessageSize_label additional.fields Unita
blockedMessages_label additional.fields Unita
data_totalThroughput_label additional.fields Unita
deliveredMessages_label additional.fields Unita
licenseEndDate_label additional.fields Unita
licenseStartDate_label additional.fields Unita
map_label additional.fields Unita
metadata_totalThroughput_label additional.fields Unita
quarantinedMessages_label additional.fields Unita
rejectedMessages_label additional.fields Unita
remainingThroughput_label additional.fields Unita
requestedMessages_label additional.fields Unita
requestedThroughput_label additional.fields Unita
sentMessages_label additional.fields Unita
throughputForecast_label additional.fields Unita
throughputLimit_label additional.fields Unita
throughput_label additional.fields Unita
totalMessages_label additional.fields Unita
undeliveredMessages_label additional.fields Unita
data.name metadata.description Mappato direttamente
desc metadata.description Mappato direttamente
data.date metadata.event_timestamp Analizzato come yyyy-MM-dd
event_type metadata.event_type Mappato direttamente
fromEnvelope network.email.bounce_address Mappato direttamente
fromHeader network.email.from Mappato direttamente
applicationName principal.administrative_domain Mappato direttamente
principal_host principal.asset.hostname Mappato direttamente
principal_host principal.hostname Mappato direttamente
principal_port principal.port Mappato direttamente
data.relayUserId principal.user.product_object_id Mappato direttamente
userId principal.user.product_object_id Mappato direttamente
applicationUserName principal.user.user_display_name Mappato direttamente
senderName target.administrative_domain Mappato direttamente
senderId target.user.product_object_id Mappato direttamente
N/D metadata.product_name Costante: PROOFPOINT SER
N/D metadata.vendor_name Costante: PROOFPOINT

Log delle modifiche

Visualizza il log delle modifiche per questo parser

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.