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 de transação e notificação. O SER gera relatórios sobre esse tráfego usando a API Reporting, uma das APIs de configuração do SER. A API retorna números agregados de uso: a posição de capacidade de processamento licenciada, o volume diário e as contagens de mensagens, além de contadores para cada usuário de retransmissão. 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 retransmissão.

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 Cloud 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 ou crie um projeto.
  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 próprio console da SER, por isso não aparece em nenhuma das entradas de navegação da 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 Secure Email Relay como o produto.

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

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

Verifique as permissões

O serviço Gerenciamento de chaves de API é compartilhado entre vários produtos da Proofpoint. Por isso, você acessa esse serviço pelo seletor de apps, e não pelo menu de navegação da SER. Para acessar as APIs de configuração da SER, incluindo a API Reporting, selecione Secure Email Relay como o produto ao criar sua chave.

Se Gerenciamento de chaves de API não aparecer no seletor de apps, sua conta não tem a permissão de gerenciamento de API. Entre em contato com o admin do Proofpoint ou com a equipe da conta para receber 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 de acesso 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 cujo 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 de funções do Cloud.
  6. Clique em Continuar.
  7. Clique em Concluído.

Essas funções são necessárias 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 permissões de gravação à conta de serviç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 que corresponde ao seu bucket do Cloud Storage (por exemplo, us-central1).
    Ambiente de execução Selecione Python 3.12 ou uma versão mais recente.

  5. Na seção Gatilho (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. Confira 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 do retransmissor solicitados por página
    MAX_PAGES 200 Limite de página para paginação de usuários de retransmissão.
    REPORT_DAY_OFFSET 1 Gerar relatório no dia de hoje menos este número de dias
    MAX_BACKFILL_DAYS 7 Na maioria dos dias, uma única execução vai fazer o backfill
    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 02:00)
    Fuso horário Selecionar 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 civil, então um dia concluído é coletado uma vez. Execute o job diariamente, depois que o dia de geração de relatórios for encerrado no fuso horário do seu locatário:

Frequência Expressão Cron Caso de uso
Diariamente às 02h 0 2 * * * Padrão (recomendado)
Diariamente às 06:00 0 6 * * * Locatários cujos relatórios são resolvidos depois
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:

    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 token está errado ou expirou. invalid_client significa que o par foi rejeitado. Crie uma chave de substituição.
  • HTTP 401 da API Reporting: o token de autenticação 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. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

  1. Acesse Configurações do SIEM > 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 Receber conta de serviço.

  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: exclui arquivos e diretórios vazios da origem após a conclusão de uma busca bem-sucedida.

    • Idade máxima do arquivo: inclua arquivos modificados no último número de dias (o padrão é 180 dias).

    • Namespace do recurso: o namespace do recurso

    • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed

  12. Clique em Próxima.

  13. Revise a nova configuração do 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 duas funções no seu bucket do Cloud Storage: Leitor de objetos do Storage para ler os objetos de registro e uma função no nível do bucket para ler os metadados dele.

  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: 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, escolha Gravador de bucket legados 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 do 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

Ver o registro de alterações deste analisador

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