Coletar registros de análise forense do Proofpoint TAP
Este documento explica como ingerir registros de análise forense do Proofpoint TAP no Google Security Operations usando o Google Cloud Storage V2.
O Proofpoint Targeted Attack Protection (TAP) é uma plataforma avançada de segurança de e-mail que detecta, analisa e bloqueia ameaças enviadas por e-mail, incluindo anexos e URLs maliciosos. A API TAP Forensics fornece evidências forenses detalhadas sobre ameaças e campanhas individuais observadas no seu ambiente, incluindo resultados de análise de sandbox, indicadores comportamentais, atividade de rede, alterações no sistema de arquivos e dados de execução de processos. Esses indicadores forenses podem ser usados para confirmar a violação em um host, enriquecer fontes de inteligência de segurança ou orquestrar atualizações em endpoints de segurança.
Antes de começar
Verifique se você tem os pré-requisitos a seguir:
- Uma instância do Google SecOps
- Um projeto do GCP com a API Storage do Cloud ativada
- Permissões para criar e gerenciar intervalos do GCS
- Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
- Uma assinatura do Proofpoint TAP com acesso ao painel de insights sobre ameaças
- Credenciais de serviço da API TAP (entidade de serviço e secret) com permissões para acessar as APIs SIEM e Forensics
Gerar credenciais de serviço da API TAP do Proofpoint
- Faça login no painel de insights de ameaças do Proofpoint TAP.
- Acesse Configurações > Aplicativos conectados > Credenciais de serviço.
- Clique em Criar nova credencial.
Na caixa de diálogo Credencial de serviço gerada, copie e armazene com segurança:
- Principal de serviço: o identificador principal usado para autenticação de API.
- Secret: a chave secreta usada para autenticação de API.
Verificar o acesso à API
Teste suas credenciais antes de prosseguir com a integração:
PRINCIPAL="your-service-principal" SECRET="your-secret" # Test SIEM API access (fetch last 5 minutes of events) curl -s "https://tap-api-v2.proofpoint.com/v2/siem/all?format=json&sinceSeconds=300" \ --user "${PRINCIPAL}:${SECRET}" # Test Forensics API access (requires a valid threatId) # curl -s "https://tap-api-v2.proofpoint.com/v2/forensics?threatId=<threatId>" \ # --user "${PRINCIPAL}:${SECRET}"
Uma resposta bem-sucedida da API SIEM retorna um objeto JSON que contém matrizes messagesBlocked, messagesDelivered, clicksBlocked e clicksPermitted.
- Se você receber um erro
401, verifique se o principal de serviço e o segredo estão corretos. - Se você receber um erro
403, confirme se sua conta tem acesso à API TAP ativado.
Criar um bucket do Google Cloud Storage
- Acesse o Console do Google Cloud.
- Selecione ou crie um projeto.
- 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 globalmente exclusivo, por exemplo, proofpoint-tap-forensics-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 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 a 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
tap-forensics-collector-sa - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect Proofpoint TAP Forensics logs
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
- Clique em Selecionar papel.
- Pesquise e selecione Administrador de objetos do Storage.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Run.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador de funções do Cloud.
- Clique em Continuar.
- Clique em Concluído.
Essas funções são necessárias para:
- Administrador de objetos do Storage: grava dados de evidências forenses no 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,
tap-forensics-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Atribuir papéis: selecione Administrador de objetos do Storage.
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
- 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.
- 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
tap-forensics-collector-trigger - Não mude as outras configurações.
- ID do tópico: insira
- Clique em Criar.
Criar uma função do Cloud Run para coletar evidências forenses
A função do Cloud Run será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar eventos de ameaça da API SIEM do Proofpoint TAP, recuperar evidências forenses de cada ameaça exclusiva usando a API Forensics e gravar os resultados 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 tap-forensics-collectorRegião Selecione a região correspondente 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 Gatilho (opcional):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico
tap-forensics-collector-trigger. - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Confira o Identity and Access Management (IAM).
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione a conta de serviço
tap-forensics-collector-sa.
- Conta de serviço: selecione a conta de serviço
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_BUCKETproofpoint-tap-forensics-logsNome do bucket do GCS GCS_PREFIXtap-forensicsPrefixo para arquivos de registro STATE_KEYtap-forensics/state.jsonCaminho do arquivo de estado TAP_PRINCIPALyour-service-principalPrincipal de serviço da API TAP TAP_SECRETyour-secretChave secreta da API TAP LOOKBACK_HOURS1Período inicial de retorno em horas (máximo de 7 dias) MAX_THREATS500Número máximo de ameaças exclusivas para buscar perícias por execução Role a tela para baixo na seção Variáveis e secrets até Solicitações:
- Tempo limite da solicitação: insira
540segundos (9 minutos)
- Tempo limite da solicitação: insira
Acesse a guia Configurações:
- Na seção Recursos:
- Memória: selecione 512 MiB ou mais.
- CPU: selecione 1
- Na seção Recursos:
Na seção Escalonamento de revisão:
- Número mínimo de instâncias: insira
0 - Número máximo de instâncias: insira
100ou ajuste com base na carga esperada.
- Número mínimo de instâncias: insira
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço for criado, o editor de código inline será aberto automaticamente.
Adicionar código da função
- Insira main no campo Ponto de entrada.
No editor de código inline, 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, timedelta import time import base64 # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.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', 'tap-forensics').strip('/') STATE_KEY = os.environ.get('STATE_KEY') or f"{GCS_PREFIX}/state.json" TAP_PRINCIPAL = os.environ.get('TAP_PRINCIPAL') TAP_SECRET = os.environ.get('TAP_SECRET') LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '1')) MAX_THREATS = int(os.environ.get('MAX_THREATS', '500')) API_BASE = 'https://tap-api-v2.proofpoint.com' def get_auth_header(): """Build HTTP Basic Authentication header.""" auth_string = f"{TAP_PRINCIPAL}:{TAP_SECRET}" auth_bytes = auth_string.encode('utf-8') auth_b64 = base64.b64encode(auth_bytes).decode('utf-8') return f"Basic {auth_b64}" @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Proofpoint TAP forensic evidence and write to GCS. The function first queries the SIEM API to discover threat IDs, then calls the Forensics API for each unique threat to retrieve detailed forensic evidence (sandbox results, behavioral indicators, network activity, file changes, and process data). Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, TAP_PRINCIPAL, TAP_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) last_time = None if isinstance(state, dict) and state.get('last_event_time'): try: last_time = parse_datetime(state['last_event_time']) # Overlap by 2 minutes to catch delayed events last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) # TAP SIEM API allows max 1 hour per request and max 7 days lookback if (now - last_time) > timedelta(days=7): last_time = now - timedelta(days=7) print("Warning: Lookback capped to 7 days (TAP API limit)") print(f"Fetching threats from {last_time.isoformat()} to {now.isoformat()}") # Step 1: Fetch threat IDs from the SIEM API threat_ids = fetch_threat_ids(last_time, now) if not threat_ids: print("No threats found in the specified time window.") save_state(bucket, STATE_KEY, now.isoformat()) return print(f"Found {len(threat_ids)} unique threat IDs") # Step 2: Fetch forensic evidence for each threat forensic_records = fetch_forensics_for_threats(threat_ids) if not forensic_records: print("No forensic evidence retrieved.") save_state(bucket, STATE_KEY, now.isoformat()) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/tap_forensics_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(record, ensure_ascii=False) for record in forensic_records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(forensic_records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed forensics for {len(threat_ids)} threats") except Exception as e: print(f'Error processing TAP forensics: {str(e)}') raise def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, key, last_event_time_iso): """Save the last event timestamp to GCS state file.""" try: state = {'last_event_time': last_event_time_iso} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_event_time={last_event_time_iso}") except Exception as e: print(f"Warning: Could not save state: {e}") def parse_datetime(value): """Parse ISO datetime string to datetime object.""" if value.endswith('Z'): value = value[:-1] + '+00:00' return datetime.fromisoformat(value) def fetch_threat_ids(start_time, end_time): """ Fetch unique threat IDs from the TAP SIEM API by querying in 1-hour intervals within the specified time window. Args: start_time: Start of the time window (datetime) end_time: End of the time window (datetime) Returns: Set of unique threat ID strings """ headers = { 'Authorization': get_auth_header(), 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-TAPForensicsCollector/1.0', } threat_ids = set() current_start = start_time backoff = 1.0 while current_start < end_time: # TAP SIEM API allows max 1 hour per request current_end = min(current_start + timedelta(hours=1), end_time) interval = ( f"{current_start.strftime('%Y-%m-%dT%H:%M:%SZ')}" f"/{current_end.strftime('%Y-%m-%dT%H:%M:%SZ')}" ) url = f"{API_BASE}/v2/siem/all?format=json&interval={interval}" try: response = http.request('GET', url, headers=headers) if response.status == 429: retry_after = int( response.headers.get('Retry-After', str(int(backoff))) ) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 60.0) continue backoff = 1.0 if response.status != 200: print(f"SIEM API HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text[:500]}") current_start = current_end continue data = json.loads(response.data.decode('utf-8')) # Extract threat IDs from all event types for key in ['messagesBlocked', 'messagesDelivered']: for msg in data.get(key, []): for threat_info in msg.get('threatsInfoMap', []): tid = threat_info.get('threatID') if tid: threat_ids.add(tid) for key in ['clicksBlocked', 'clicksPermitted']: for click in data.get(key, []): tid = click.get('threatID') if tid: threat_ids.add(tid) event_count = sum( len(data.get(k, [])) for k in [ 'messagesBlocked', 'messagesDelivered', 'clicksBlocked', 'clicksPermitted', ] ) print( f"Interval {interval}: {event_count} events, " f"{len(threat_ids)} unique threats so far" ) except Exception as e: print(f"Error fetching SIEM events: {e}") current_start = current_end if len(threat_ids) >= MAX_THREATS: print(f"Reached max threats limit ({MAX_THREATS})") break return threat_ids def fetch_forensics_for_threats(threat_ids): """ Fetch forensic evidence for each threat ID from the Forensics API. Args: threat_ids: Set of threat ID strings Returns: List of forensic report dictionaries """ headers = { 'Authorization': get_auth_header(), 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-TAPForensicsCollector/1.0', } records = [] backoff = 1.0 processed = 0 skipped = 0 for threat_id in list(threat_ids)[:MAX_THREATS]: url = ( f"{API_BASE}/v2/forensics" f"?threatId={threat_id}" f"&includeCampaignForensics=true" ) try: response = http.request('GET', url, headers=headers) if response.status == 429: retry_after = int( response.headers.get('Retry-After', str(int(backoff))) ) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 60.0) # Retry the same threat continue backoff = 1.0 if response.status == 204: skipped += 1 processed += 1 continue if response.status != 200: print( f"Forensics API error for {threat_id}: " f"HTTP {response.status}" ) skipped += 1 processed += 1 continue data = json.loads(response.data.decode('utf-8')) reports = data.get('reports', []) if reports: # Add the threat ID to each report for correlation for report in reports: report['_threatId'] = threat_id records.extend(reports) processed += 1 if processed % 50 == 0: print( f"Progress: {processed}/{len(threat_ids)} threats, " f"{len(records)} forensic reports collected" ) except Exception as e: print(f"Error fetching forensics for {threat_id}: {e}") skipped += 1 processed += 1 print( f"Forensics collection complete: {processed} threats processed, " f"{skipped} skipped, {len(records)} reports collected" ) return recordsSegundo 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).
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.
- No console do GCP, acesse o Cloud Scheduler.
- Clique em Criar job.
Informe os seguintes detalhes de configuração:
Configuração Valor Nome tap-forensics-collector-hourlyRegião Selecione a mesma região da função do Cloud Run Frequência 0 * * * *(de hora em hora)Fuso horário Selecionar fuso horário (UTC recomendado) Tipo de destino Pub/Sub Tópico Selecione o tópico tap-forensics-collector-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 15 minutos | */15 * * * * |
Ambientes de alto volume com muitas ameaças |
| A cada hora | 0 * * * * |
Padrão (recomendado) |
| A cada 6 horas | 0 */6 * * * |
Ambientes de baixo volume |
Testar a integração
- No console do Cloud Scheduler, encontre seu job.
- Clique em Forçar execução para acionar o job manualmente.
- Aguarde alguns segundos.
- Acesse Cloud Run > Serviços.
- Clique no nome da função
tap-forensics-collector. - Clique na guia Registros.
Verifique se a função foi executada com sucesso. Procure:
Fetching threats from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Interval .../...: X events, Y unique threats so far Found Z unique threat IDs Forensics collection complete: Z threats processed, 0 skipped, N reports collected Wrote N records to gs://proofpoint-tap-forensics-logs/tap-forensics/tap_forensics_YYYYMMDD_HHMMSS.ndjson Successfully processed forensics for Z threatsAcesse Cloud Storage > Buckets.
Clique no nome do bucket.
Navegue até a pasta de prefixo
tap-forensics/.Verifique se um novo arquivo
.ndjsonfoi criado com o carimbo de data/hora atual.
Se você encontrar erros nos registros:
- HTTP 401: verifique TAP_PRINCIPAL e TAP_SECRET nas variáveis de ambiente. Verifique se o Service Principal e a chave secreta estão corretos.
- HTTP 403: confirme se o acesso à API está ativado na sua conta do TAP.
- HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera. Considere reduzir a frequência da programação.
- Nenhuma ameaça encontrada: isso é normal se nenhuma ameaça foi detectada no período. O TAP só informa ameaças identificadas pela defesa de URL ou de anexos.
- Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.
Configurar um feed no Google SecOps para ingerir registros de análise forense do Proofpoint TAP
- 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,
Proofpoint TAP Forensics). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Proofpoint Tap Forensics como o Tipo de registro.
Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopie esse endereço de e-mail para usar na próxima etapa.
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://proofpoint-tap-forensics-logs/tap-forensics/- Substitua:
proofpoint-tap-forensics-logs: o nome do bucket do GCS.tap-forensics: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
- Substitua:
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.
- Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência.
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
Clique em Próxima.
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 do papel 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 principais: cole o e-mail da conta de serviço do Google SecOps.
- Atribuir papéis: selecione Leitor de objetos do Storage.
Clique em Salvar.
Tabela de mapeamento de UDM
| Campo de registro | Mapeamento do UDM | Lógica |
|---|---|---|
malicious_label |
additional.fields |
Mesclado |
threat_type_label |
additional.fields |
Mesclado |
generated |
metadata.event_timestamp |
Analisado como ISO8601 |
has_principal |
metadata.event_type |
Mapeado: true → STATUS_UPDATE |
protocol |
network.ip_protocol |
Mapeado diretamente |
prin_ip |
principal.asset.ip |
Mesclado |
domain |
principal.domain.name |
Mapeado diretamente |
path |
principal.file.full_path |
Mapeado diretamente |
file_hash_sha256 |
principal.file.sha256 |
Mapeado diretamente |
prin_ip |
principal.ip |
Mesclado |
port |
principal.port |
Mapeado diretamente |
url |
principal.url |
Mapeado diretamente |
_security_result |
security_result |
Mesclado |
| N/A | metadata.event_type |
Constante: STATUS_UPDATE |
| N/A | metadata.product_event_type |
Constante: Forensic Reports |
| N/A | metadata.product_name |
Constante: TAP Forensics |
| N/A | metadata.vendor_name |
Constante: Proofpoint |
| N/A | principal.asset.platform_software.platform |
Constante: WINDOWS |
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.