Coletar registros de IOCs do Proofpoint Emerging Threats Pro

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros de IOC do Proofpoint Emerging Threats Pro no Google Security Operations usando o Google Cloud Storage. A Emerging Threats Intelligence publica listas de reputação por hora para IPs e domínios em formato CSV com dados de inteligência contra ameaças, incluindo categorias, pontuações e informações temporais. O código do analisador processa dados de inteligência de ameaças ET_PRO formatados em CSV. Ele extrai endereços IP, domínios, categorias, pontuações e outras informações relevantes, mapeando-os para um formato de IOC padronizado e para o esquema UDM do Chronicle para análise e uso posteriores no Google SecOps.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

Uma instância do Google SecOps
Um projeto do GCP com a API Cloud Storage ativada
Permissões para criar e gerenciar buckets do GCS
Permissões para gerenciar políticas do IAM em buckets do GCS
Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
Assinatura da Proofpoint ET Intelligence com acesso a listas de reputação.
Chave de API da ET Intelligence em https://etadmin.proofpoint.com/api-access

Reúna os pré-requisitos do Emerging Threats Pro

Faça login no portal de administração da ET Intelligence em https://etadmin.proofpoint.com.
Acesse Acesso à API.
Copie e salve sua chave de API.
Entre em contato com seu representante da Proofpoint para receber:
- URL da lista detalhada de reputação de IP
- URL da lista detalhada de reputação de domínio

A ET Intelligence fornece arquivos CSV separados para listas de reputação de IP e domínio, atualizadas a cada hora. Use o formato "detalhado", que inclui estas colunas:

Lista de domínios: nome de domínio, categoria, pontuação, primeira vez visto, última vez visto, portas
Lista de IPs: endereço IP, categoria, pontuação, primeira vez visto, última vez visto, portas

Os URLs de formato detalhado geralmente seguem este padrão:

Lista de IPs: https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txt
Lista de domínios: https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txt

O cabeçalho de autorização precisa conter o valor bruto da chave de API sem um prefixo "Bearer", de acordo com as expectativas da API Emerging Threats.

Criar um bucket do Google Cloud Storage

Acesse o Console do Google Cloud.
Selecione seu projeto ou crie um novo.
No menu de navegação, acesse Cloud Storage > Buckets.
Clique em Criar bucket.

Informe os seguintes detalhes de configuração:

Configuração	Valor
Nomeie seu bucket	Insira um nome exclusivo globalmente, por exemplo, `et-pro-ioc-bucket`.
Tipo de local	Escolha com base nas 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

Clique em Criar.

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 no bucket do GCS e ser invocada pelo Pub/Sub.

Criar conta de serviço

No Console do GCP, acesse IAM e administrador > Contas de serviço.
Clique em Criar conta de serviço.
Informe os seguintes detalhes de configuração:
- Nome da conta de serviço: insira et-pro-ioc-fetcher-sa.
- Descrição da conta de serviço: insira Service account for Cloud Run function to collect Proofpoint ET Pro IOC logs.
Clique em Criar e continuar.
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.
Clique em Continuar.
Clique em Concluído.

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

Administrador de objetos do Storage: grava registros em um bucket do GCS 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 GCS

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

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Acesse a guia Permissões.
Clique em Conceder acesso.
Informe os seguintes detalhes de configuração:
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo, et-pro-ioc-fetcher-sa@PROJECT_ID.iam.gserviceaccount.com).
- Atribuir papéis: selecione Administrador de objetos do Storage.
Clique em Salvar.

Criar tópico 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.

No Console do GCP, acesse Pub/Sub > Tópicos.
Selecione Criar tópico.
Informe os seguintes detalhes de configuração:
- ID do tópico: insira et-pro-ioc-trigger.
- Não altere as outras configurações.
Clique em Criar.

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

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API Proofpoint ET Intelligence e gravá-los no GCS.

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

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

Configuração	Valor
Nome do serviço	`et-pro-ioc-fetcher`
Região	Selecione a região que corresponde ao seu bucket do GCS (por exemplo, `us-central1`).
Ambiente de execução	Selecione Python 3.12 ou uma versão mais recente.

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 et-pro-ioc-trigger.
4. Clique em Salvar.
Na seção Autenticação:
1. Selecione Exigir autenticação.
2. Confira o Identity and Access Management (IAM).
Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione et-pro-ioc-fetcher-sa.

Acesse a guia Contêineres:

Clique em Variáveis e secrets.
Clique em + Adicionar variável para cada variável de ambiente:

Nome da variável	Valor de exemplo	Descrição
`GCS_BUCKET`	`et-pro-ioc-bucket`	Nome do bucket do GCS
`GCS_PREFIX`	`et-pro-ioc`	Prefixo para arquivos de registro
`STATE_KEY`	`et-pro-ioc/state.json`	Caminho do arquivo de estado
`ET_API_KEY`	`your-et-api-key`	Chave da API ET Intelligence
`ET_IP_LIST_URL`	`https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txt`	URL da lista detalhada de reputação de IP
`ET_DOMAIN_LIST_URL`	`https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txt`	URL da lista detalhada de reputação de domínio
`TIMEOUT`	`120`	Tempo limite da solicitação HTTP em segundos

Role a tela para baixo na guia Variáveis e secrets até Solicitações:
- Tempo limite da solicitação: insira 600 segundos (10 minutos).
Acesse a guia Configurações em Contêineres:
- Na seção Recursos:
  - Memória: selecione 512 MiB ou mais.
  - CPU: selecione 1.
- Clique em Concluído.
Role até Ambiente de execução:
- Selecione Padrão (recomendado).
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.
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

Insira main em Ponto de entrada da função.

No editor de código em linha, crie dois arquivos:

Primeiro arquivo: main.py::

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone
import time

# 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()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch ET Pro IOC reputation lists and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'et-pro-ioc').strip('/')
    state_key = os.environ.get('STATE_KEY', f'{prefix}/state.json')
    et_api_key = os.environ.get('ET_API_KEY')
    et_ip_list_url = os.environ.get('ET_IP_LIST_URL')
    et_domain_list_url = os.environ.get('ET_DOMAIN_LIST_URL')
    timeout = int(os.environ.get('TIMEOUT', '120'))

    if not all([bucket_name, et_api_key, et_ip_list_url, et_domain_list_url]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        # Generate timestamp for file naming
        now = datetime.now(timezone.utc)
        timestamp = now.strftime('%Y/%m/%d/%H%M%S')

        results = []
        errors = []

        # Fetch IP reputation list
        try:
            print('Fetching IP reputation list...')
            ip_data = fetch_with_retry(et_ip_list_url, et_api_key, timeout)
            ip_key = f'{prefix}/ip/{timestamp}.csv'
            save_to_gcs(bucket, ip_key, ip_data)
            results.append({'type': 'ip', 'key': ip_key, 'size': len(ip_data)})
            print(f'Successfully fetched IP list: {len(ip_data)} bytes')
        except Exception as e:
            error_msg = f'Failed to fetch IP list: {str(e)}'
            print(error_msg)
            errors.append(error_msg)

        # Fetch Domain reputation list
        try:
            print('Fetching Domain reputation list...')
            domain_data = fetch_with_retry(et_domain_list_url, et_api_key, timeout)
            domain_key = f'{prefix}/domain/{timestamp}.csv'
            save_to_gcs(bucket, domain_key, domain_data)
            results.append({'type': 'domain', 'key': domain_key, 'size': len(domain_data)})
            print(f'Successfully fetched Domain list: {len(domain_data)} bytes')
        except Exception as e:
            error_msg = f'Failed to fetch Domain list: {str(e)}'
            print(error_msg)
            errors.append(error_msg)

        # Save state
        state = {
            'last_fetch': now.isoformat(),
            'results': results,
            'errors': errors
        }
        save_state(bucket, state_key, state)

        if errors:
            print(f'Completed with {len(errors)} error(s)')
        else:
            print('Successfully completed all fetches')

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

def fetch_with_retry(url, api_key, timeout, max_retries=3):
    """Fetch URL with retry logic for rate limits."""
    if not url.lower().startswith('https://'):
        raise ValueError('Only HTTPS URLs are allowed')

    headers = {'Authorization': api_key}

    for attempt in range(max_retries):
        try:
            response = http.request('GET', url, headers=headers, timeout=timeout)

            if response.status == 200:
                return response.data
            elif response.status == 429:
                # Rate limited, wait and retry
                wait_time = min(30 * (2 ** attempt), 300)
                print(f'Rate limited, waiting {wait_time}s...')
                time.sleep(wait_time)
            else:
                raise Exception(f'HTTP {response.status}: {response.reason}')
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(5 * (attempt + 1))

    raise Exception(f'Failed to fetch {url} after {max_retries} attempts')

def save_to_gcs(bucket, key, content):
    """Save content to GCS with appropriate content type."""
    blob = bucket.blob(key)
    blob.upload_from_string(content, content_type='text/csv')
    print(f'Saved {len(content)} bytes to gs://{bucket.name}/{key}')

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

Segundo arquivo: requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).

Observação: a configuração do acionador do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.

Criar o job do Cloud Scheduler

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

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

Informe os seguintes detalhes de configuração:

Configuração	Valor
Nome	`et-pro-ioc-fetcher-hourly`
Região	Selecione a mesma região da função do Cloud Run
Frequência	`0 * * * *` (a cada hora, na hora)
Fuso horário	Selecione o fuso horário (UTC recomendado)
Tipo de destino	Pub/Sub
Tópico	Selecionar `et-pro-ioc-trigger`
Corpo da mensagem	`{}` (objeto JSON vazio)

Clique em Criar.

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

Escolha a frequência com base no volume de registros e nos requisitos de latência:

Frequência	Expressão Cron	Caso de uso
A cada hora	`0 * * * *`	Padrão (recomendado para IOC do ET Pro)
A cada 2 horas	`0 /2 * *`	Frequência mais baixa
A cada 6 horas	`0 /6 * *`	Atualizações mínimas

Testar a integração

No console do Cloud Scheduler, encontre seu job.
Clique em Executar à força para acionar o job manualmente.
Aguarde alguns segundos.
Acesse Cloud Run > Serviços.
Clique no nome da função (et-pro-ioc-fetcher).
Clique na guia Registros.

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

Fetching IP reputation list...
Successfully fetched IP list: X bytes
Fetching Domain reputation list...
Successfully fetched Domain list: X bytes
Successfully completed all fetches

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Navegue até as pastas de prefixo (et-pro-ioc/ip/ e et-pro-ioc/domain/).
Verifique se os novos arquivos .csv foram criados com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

HTTP 401: verifique ET_API_KEY nas variáveis de ambiente
HTTP 403: verifique se a chave de API tem as permissões necessárias
HTTP 429: limitação de taxa. A função vai tentar novamente automaticamente com espera.
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 GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

Acesse Configurações do SIEM > Feeds.
Clique em Adicionar novo feed.
Clique em Configurar um único feed.
No campo Nome do feed, insira um nome para o feed (por exemplo, ET Pro IOC - IP Reputation).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Emerging Threats Pro como o Tipo de registro.
Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
Copie esse endereço de e-mail para usar na próxima etapa.

Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.

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

A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Acesse a guia Permissões.
Clique em Conceder acesso.
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 Leitor de objetos do Storage.
Clique em Salvar.

Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar feeds no Google SecOps para ingerir registros de IOCs do Proofpoint Emerging Threats Pro

É necessário criar dois feeds separados: um para reputação de IP e outro para reputação de domínio.

Criar um feed de reputação de IP

Acesse Configurações do SIEM > Feeds.
Clique em Adicionar novo feed.
Clique em Configurar um único feed.
No campo Nome do feed, insira ET Pro IOC - IP Reputation.
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Emerging Threats Pro como o Tipo de registro.
Clique em Próxima.
Especifique valores para os seguintes parâmetros de entrada:
- URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:
```
gs://et-pro-ioc-bucket/et-pro-ioc/ip/
```
  - Substitua et-pro-ioc-bucket pelo nome real do bucket do GCS.
  Observação: sempre inclua a barra (/) no final do URI.
- Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:
  - Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
  - Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.
  - Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.
    
    Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.
- Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
- Namespace do recurso: o namespace do recurso.
- Rótulos de ingestão: o rótulo aplicado aos eventos deste feed.
Clique em Próxima.
Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Criar feed de reputação de domínio

Repita o processo de criação do feed:

Acesse Configurações do SIEM > Feeds.
Clique em Adicionar novo feed.
Clique em Configurar um único feed.
No campo Nome do feed, insira ET Pro IOC - Domain Reputation.
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Emerging Threats Pro como o Tipo de registro.
Clique em Próxima.
Especifique valores para os seguintes parâmetros de entrada:
- URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:
```
gs://et-pro-ioc-bucket/et-pro-ioc/domain/
```
Substitua et-pro-ioc-bucket pelo nome real do bucket do GCS.

Observação: sempre inclua a barra (/) no final do URI.
- Opção de exclusão de fonte: selecione de acordo com sua preferência.
- Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
- Namespace do recurso: o namespace do recurso.
- Rótulos de ingestão: o rótulo aplicado aos eventos deste feed.
Clique em Próxima.
Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Tabela de mapeamento do UDM

Campo de registro	Mapeamento do UDM	Lógica
categoria	Esse campo é usado na lógica do analisador, mas não é mapeado diretamente para a UDM.	Ele determina o valor de event.ioc.categorization usando uma tabela de pesquisa.
collection_time.nanos	event.idm.entity.metadata.collected_timestamp.nanos	Mapeado diretamente do registro bruto.
collection_time.seconds	event.idm.entity.metadata.collected_timestamp.seconds	Mapeado diretamente do registro bruto.
dados	Esse campo é analisado em vários campos da UDM com base no conteúdo dele.
first_seen	event.idm.entity.metadata.interval.start_time	Analisado como uma data e mapeado para o UDM.
first_seen	event.ioc.active_timerange.start	Analisado como uma data e mapeado para o UDM.
ip_or_domain	event.idm.entity.entity.hostname	Mapeado para o UDM se o padrão grok extrair um host do campo.
ip_or_domain	event.idm.entity.entity.ip	Mapeado para o UDM se o padrão grok não extrair um host do campo.
ip_or_domain	event.ioc.domain_and_ports.domain	Mapeado para o UDM se o padrão grok extrair um host do campo.
ip_or_domain	event.ioc.ip_and_ports.ip_address	Mapeado para o UDM se o padrão grok não extrair um host do campo.
last_seen	event.idm.entity.metadata.interval.end_time	Analisado como uma data e mapeado para o UDM.
last_seen	event.ioc.active_timerange.end	Analisado como uma data e mapeado para o UDM.
ports	event.idm.entity.entity.labels.value	Analisado, unido com delimitador de vírgula e mapeado para a UDM se houver várias portas.
ports	event.idm.entity.entity.port	Analisado e mapeado para a UDM se houver apenas uma porta.
ports	event.ioc.domain_and_ports.ports	Analisado e mapeado para o UDM se o padrão grok extrair um host do campo.
ports	event.ioc.ip_and_ports.ports	Analisado e mapeado para o UDM se o padrão grok não extrair um host do campo.
score	event.ioc.confidence_score	Mapeado diretamente do registro bruto.
	event.idm.entity.entity.labels.key	Defina como "ports" se houver várias portas.
	event.idm.entity.metadata.entity_type	Definido como "DOMAIN_NAME" se o padrão grok extrair um host do campo "ip_or_domain". Caso contrário, será definido como "IP_ADDRESS".
	event.idm.entity.metadata.threat.category	Definido como "SOFTWARE_MALICIOUS".
	event.idm.entity.metadata.threat.category_details	Derivado do campo "Categoria" usando uma tabela de pesquisa.
	event.idm.entity.metadata.threat.threat_name	Defina como "ET Intelligence Rep List".
	event.idm.entity.metadata.vendor_name	Defina como "ET_PRO_IOC".
	event.ioc.feed_name	Defina como "ET Intelligence Rep List".
	event.ioc.raw_severity	Definido como "Malicioso".
	timestamp.nanos	Copiado de "collection_time.nanos".
	timestamp.seconds	Copiado de "collection_time.seconds".

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