Coletar registros do Proofpoint Secure Email Relay

Compatível com:

Este documento explica como ingerir registros do Proofpoint Secure Email Relay no Google Security Operations usando o Cloud Storage V2.

O Proofpoint Secure Email Relay (SER) é um redirecionamento gerenciado para e-mails gerados por aplicativos, como mensagens transacionais e de notificação. O SER informa sobre esse tráfego usando a API Reporting, uma das APIs de configuração do SER. A API retorna números de uso agregados: a posição de capacidade licenciada, o volume diário e as contagens de mensagens, além de contadores para cada usuário de redirecionamento. O analisador mapeia esses números para o modelo de dados unificado (UDM), registrando-os como rótulos ao lado da identidade do usuário de redirecionamento.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Uma instância do Google SecOps
  • Um projeto do Google Cloud com a API Storage do Cloud Storage ativada.
  • Permissões para criar e gerenciar buckets do Cloud Storage
  • Permissões para gerenciar políticas do Identity and Access Management (IAM) em buckets do Cloud Storage
  • Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Acesso de administrador ao portal do Proofpoint Secure Email Relay, incluindo permissão para criar chaves de API

Criar um bucket do Cloud Storage

  1. Acesse o console doGoogle Cloud .
  2. Selecione seu projeto ou crie um novo.
  3. No menu de navegação, acesse Cloud Storage > Buckets.
  4. Clique em Criar bucket.

  5. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nomeie seu bucket Insira um nome globalmente exclusivo (por exemplo, proofpoint-ser-logs)
    Tipo de local Escolha de acordo com suas necessidades (região, birregional, multirregional)
    Local Selecione o local (por exemplo, us-central1).
    Classe de armazenamento Padrão (recomendado para registros acessados com frequência)
    Controle de acesso Uniforme (recomendado)
    Ferramentas de proteção Opcional: ativar o controle de versões de objetos ou a política de retenção

  6. Clique em Criar.

Coletar credenciais da API SER do Proofpoint

O acesso à API Reporting é concedido por uma chave de API emitida pelo serviço de gerenciamento de chaves de API da Proofpoint. A chave não é criada no console do SER, por isso não aparece em nenhuma das entradas de navegação do SER.

Criar uma chave de API

  1. Faça login no portal do Proofpoint Secure Email Relay com as credenciais de administrador.
  2. Abra o seletor de apps no canto superior esquerdo.
  3. Acesse Serviços > Gerenciamento de chaves de API.
  4. Clique em Criar chave.
  5. Selecione Retransmissão de e-mail segura como o produto.

  6. Copie e armazene com segurança os seguintes valores:

    • Chave: usada como o client_id
    • Secret: usado como o client_secret

Verifique as permissões

O serviço de gerenciamento de chaves de API é compartilhado entre vários produtos da Proofpoint. Por isso, ele é acessado pelo seletor de apps, e não pelo menu de navegação do SER. Para ter acesso às APIs de configuração do SER, incluindo a API Reporting, selecione Retransmissão de e-mail segura como o produto ao criar sua chave.

Se a opção Gerenciamento de chaves de API não aparecer no seletor de apps, sua conta não tem a permissão de gerenciamento de APIs. Entre em contato com o administrador ou a equipe de conta da Proofpoint para receber a permissão.

Testar o acesso à API

  • A API Reporting usa a concessão de credenciais do cliente OAuth 2.0. Troque a chave e o secret por um token do portador e chame a API com esse 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"

Uma chamada bem-sucedida retorna um objeto JSON em que o membro data contém throughputLimit, licenseStartDate, licenseEndDate e as médias de capacidade de processamento.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar em um bucket do Cloud Storage e ser invocada pelo Pub/Sub.

Criar a conta de serviço

  1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
  2. Clique em Criar conta de serviço.
  3. Informe os seguintes detalhes de configuração:
    • Nome da conta de serviço: insira proofpoint-ser-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs.
  4. Clique em Criar e continuar.
  5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
    1. Clique em Selecionar papel.
    2. Pesquise e selecione Administrador de objetos do Storage.
    3. Clique em + Adicionar outro papel.
    4. Pesquise e selecione Invocador do Cloud Run.
    5. Clique em + Adicionar outro papel.
    6. Pesquise e selecione Invocador do Cloud Functions.
  6. Clique em Continuar.
  7. Clique em Concluído.

Esses papéis são necessários para:

  • Administrador de objetos do Storage: grava registros em um bucket do Cloud Storage e gerencia arquivos de estado.
  • Invocador do Cloud Run: permite que o Pub/Sub invoque a função.
  • Invocador do Cloud Functions: permite a invocação de funções.

Conceder permissões do IAM no bucket do Cloud Storage

Conceda à conta de serviço permissões de gravação no bucket do Cloud Storage:

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (por exemplo, proofpoint-ser-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • Atribuir papéis: selecione Administrador de objetos do Storage.
  6. Clique em Salvar.

Criar um tópico do Pub/Sub

Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

  1. No Console do GCP, acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.
  3. Informe os seguintes detalhes de configuração:
    • ID do tópico: insira proofpoint-ser-trigger.
    • Não mude as outras configurações.
  4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar relatórios de uso da API SER Reporting e gravá-los no Cloud Storage.

  1. No Console do GCP, acesse o Cloud Run.
  2. Clique em Criar serviço.
  3. Selecione Função (use um editor in-line para criar uma função).

  4. Na seção Configurar, forneça os seguintes detalhes de configuração:

    Configuração Valor
    Nome do serviço proofpoint-ser-collector
    Região Selecione a região correspondente ao bucket do Cloud Storage (por exemplo, us-central1)
    Ambiente de execução Selecione Python 3.12 ou mais recente.

  5. Na seção Acionador (opcional):

    1. Clique em + Adicionar gatilho.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico do Pub/Sub (proofpoint-ser-trigger).
    4. Clique em Salvar.

  6. Na seção Autenticação:

    1. Selecione Exigir autenticação.
    2. Verifique o Identity and Access Management (IAM).

  7. Acesse e abra Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione a conta de serviço (proofpoint-ser-collector-sa).

  9. Acesse a guia Contêineres:

    1. Clique em Variáveis e secrets.
    2. Clique em + Adicionar variável para cada variável de ambiente:
    Nome da variável Valor de exemplo Descrição
    GCS_BUCKET proofpoint-ser-logs Nome do bucket do Cloud Storage
    GCS_PREFIX ser-logs Prefixo para arquivos de registro
    STATE_KEY ser-logs-state.json Caminho do estado, fora do prefixo do registro
    TOKEN_URL https://auth.proofpoint.com/v1/token Endpoint do token OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com URL de base da API SER Reporting
    CLIENT_ID your-key A chave do gerenciamento de chaves de API.
    CLIENT_SECRET your-secret O Secret do gerenciamento de chaves de API
    PAGE_SIZE 50 Usuários de redirecionamento solicitados por página
    MAX_PAGES 200 Limite de páginas para paginação de usuários de redirecionamento
    REPORT_DAY_OFFSET 1 Relatório sobre o dia com essa quantidade de dias atrás
    MAX_BACKFILL_DAYS 7 A maioria dos dias em que uma única execução vai preencher
    SEEN_RETENTION_DAYS 7 Dias de chaves de eliminação de duplicação mantidas no estado

  10. Na seção Variáveis e secrets, acesse Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos)

  11. Acesse a guia Configurações:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.

  12. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 (ou ajuste com base na carga esperada)

  13. Clique em Criar.

  14. Aguarde a criação do serviço (1 a 2 minutos).

  15. Depois que o serviço for criado, o editor de código inline será aberto automaticamente.

Adicionar o código da função

  1. Insira main no campo Ponto de entrada.
  2. No editor de código inline, crie dois arquivos:

  • Primeiro arquivo: 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")}')

  • Segundo arquivo: requirements.txt::

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Clique em Implantar para salvar e implantar a função.

  2. Aguarde a conclusão da implantação (2 a 3 minutos).

Criar um job do Cloud Scheduler

O Cloud Scheduler vai publicar mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

  1. No Console do GCP, acesse o Cloud Scheduler.
  2. Clique em Criar job.

  3. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nome proofpoint-ser-collector-daily
    Região Selecione a mesma região da função do Cloud Run
    Frequência 0 2 * * * (diariamente, às 2h)
    Fuso horário Selecione o fuso horário (UTC recomendado)
    Tipo de destino Pub/Sub
    Tópico Selecione o tópico do Pub/Sub (proofpoint-ser-trigger)
    Corpo da mensagem {} (objeto JSON vazio)

  4. Clique em Criar.

Opções de frequência de programação

A API Reporting agrega por dia do calendário, então um dia concluído é coletado uma vez. Execute o job diariamente, após o encerramento do dia de geração de relatórios no fuso horário do locatário:

Frequência Expressão Cron Caso de uso
Diariamente, às 2h 0 2 * * * Padrão (recomendado)
Diariamente às 06:00 0 6 * * * Locatários com relatórios que são liquidados mais tarde
Duas vezes por dia 0 2,14 * * * Adiciona uma nova tentativa no mesmo dia se a primeira execução falhar.

Uma execução que encontra o dia já registrado é encerrada sem gravar nada. Portanto, uma execução extra custa uma chamada de API e nunca duplica dados.

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job.
  2. Clique em Forçar execução para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique no nome da função (proofpoint-ser-collector).
  6. Clique na guia Registros.

  7. Verifique se a função foi executada com sucesso. Procure por:

    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. Acesse Cloud Storage > Buckets.

  9. Clique no nome do bucket (proofpoint-ser-logs).

  10. Navegue até a pasta de prefixo (ser-logs/).

  11. Verifique se um novo arquivo .ndjson foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

  • HTTP 400 do endpoint do token: a chave ou o secret está incorreto ou a chave expirou. invalid_client significa que o par foi rejeitado. Crie uma chave de substituição.
  • HTTP 401 da API Reporting: o token do portador não foi enviado ou não é mais válido.
  • HTTP 429: limitação de taxa. A função tenta novamente com espera e falha na execução se o limite persistir, deixando o dia para a próxima execução.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do Cloud Storage. É necessário conceder a essa conta de serviço acesso ao bucket.

Encontrar o e-mail da conta de serviço

  1. Acesse Configurações > Gerenciamento de dados > Feeds. Se você estiver em uma plataforma somente SIEM, acesse Configurações > Feeds.
  2. Clique em Adicionar novo feed.
  3. Clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Proofpoint SER Logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione ProofPoint Secure Email Relay como o Tipo de registro.
  7. Clique em Get Service Account.

  8. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:

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

  9. Copie esse endereço de e-mail para usar na próxima etapa.

  10. Clique em Próxima.

  11. Especifique valores para os seguintes parâmetros de entrada:

    • URL do bucket de armazenamento: insira o URI do bucket do Cloud Storage com o caminho do prefixo:

      gs://proofpoint-ser-logs/ser-logs/
      • Substitua:
        • proofpoint-ser-logs: o nome do bucket do Cloud Storage.
        • ser-logs: prefixo ou caminho de pasta opcional em que os registros são armazenados (deixe em branco para a raiz).

    • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

      • Nunca excluir arquivos: nunca exclua arquivos da origem (recomendado para testes).
      • Excluir arquivos transferidos e diretórios vazios: exclua arquivos e diretórios vazios da origem após a conclusão de uma busca.

    • Idade máxima do arquivo: inclua arquivos modificados nos últimos dias (o padrão é 180 dias).

    • Namespace do recurso: o namespace do recurso

    • Rótulos de ingestão: o rótulo que será aplicado aos eventos desse feed.

  12. Clique em Próxima.

  13. Revise a configuração do novo feed na tela Finalizar e clique em Enviar.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa de dois papéis no bucket do Cloud Storage: Leitor de objetos do Storage para ler os objetos de registro e um papel no nível do bucket para ler os metadados do bucket.

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (por exemplo, proofpoint-ser-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
    • Atribuir papéis: selecione as duas opções a seguir:
      • Leitor de objetos do Storage: lê os objetos de registro.
      • Leitor de bucket legado do Storage: lê os metadados do bucket. Se você selecionou a opção de exclusão Excluir arquivos transferidos e diretórios vazios, selecione Gravador de bucket legado do Storage, que também concede a permissão de exclusão.

  6. Clique em Salvar.

Tabela de mapeamento de UDM

Campo de registro Mapeamento de UDM Lógica
acceptedMessages_label additional.fields Mesclado
acceptedThroughput_label additional.fields Mesclado
average30DayThroughput_label additional.fields Mesclado
average7DayThroughput_label additional.fields Mesclado
averageDailyThroughput_label additional.fields Mesclado
avgAcceptedMessageSize_label additional.fields Mesclado
blockedMessages_label additional.fields Mesclado
data_totalThroughput_label additional.fields Mesclado
deliveredMessages_label additional.fields Mesclado
licenseEndDate_label additional.fields Mesclado
licenseStartDate_label additional.fields Mesclado
map_label additional.fields Mesclado
metadata_totalThroughput_label additional.fields Mesclado
quarantinedMessages_label additional.fields Mesclado
rejectedMessages_label additional.fields Mesclado
remainingThroughput_label additional.fields Mesclado
requestedMessages_label additional.fields Mesclado
requestedThroughput_label additional.fields Mesclado
sentMessages_label additional.fields Mesclado
throughputForecast_label additional.fields Mesclado
throughputLimit_label additional.fields Mesclado
throughput_label additional.fields Mesclado
totalMessages_label additional.fields Mesclado
undeliveredMessages_label additional.fields Mesclado
data.name metadata.description Mapeado diretamente
desc metadata.description Mapeado diretamente
data.date metadata.event_timestamp Analisado como yyyy-MM-dd
event_type metadata.event_type Mapeado diretamente
fromEnvelope network.email.bounce_address Mapeado diretamente
fromHeader network.email.from Mapeado diretamente
applicationName principal.administrative_domain Mapeado diretamente
principal_host principal.asset.hostname Mapeado diretamente
principal_host principal.hostname Mapeado diretamente
principal_port principal.port Mapeado diretamente
data.relayUserId principal.user.product_object_id Mapeado diretamente
userId principal.user.product_object_id Mapeado diretamente
applicationUserName principal.user.user_display_name Mapeado diretamente
senderName target.administrative_domain Mapeado diretamente
senderId target.user.product_object_id Mapeado diretamente
N/A metadata.product_name Constante: PROOFPOINT SER
N/A metadata.vendor_name Constante: PROOFPOINT

Registro de alterações

Confira o registro de alterações deste analisador

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.