Coletar registros da plataforma ZeroFox
Este documento explica como ingerir registros da plataforma ZeroFox no Google Security Operations usando o Google Cloud Storage. A plataforma ZeroFox oferece proteção contra riscos digitais monitorando e analisando ameaças em mídias sociais, apps para dispositivos móveis, nuvem, e-mail e outros canais digitais.
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
- Acesso privilegiado ao locatário da plataforma ZeroFox.
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, zerofox-platform-logs.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.
Coletar credenciais da plataforma ZeroFox
Receber token de acesso pessoal do ZeroFox
- Faça login na plataforma ZeroFox em https://cloud.zerofox.com.
- Acesse Configurações > Conexões de dados > Feeds de dados da API.
- URL direto (após o login): https://cloud.zerofox.com/data_connectors/api
- Observação: se você não encontrar esse item de menu, entre em contato com o administrador do ZeroFox para ter acesso. O menu também pode ser chamado de Conectores de dados > Feeds de dados da API, dependendo da versão da interface do locatário.
- Clique em Gerar token ou Criar token de acesso pessoal.
- Informe os seguintes detalhes de configuração:
- Nome: insira um nome descritivo, por exemplo,
Google SecOps GCS Ingestion. - Expiração: selecione um período de rotação de acordo com a política de segurança da sua organização.
- Permissões/Feeds: selecione permissões de leitura para:
- Alertas
- Feeds de CTI
- Outros tipos de dados que você quer exportar
- Nome: insira um nome descritivo, por exemplo,
- Clique em Gerar.
- Copie e salve o token de acesso pessoal gerado em um local seguro. Ele não poderá ser acessado novamente.
Salve o ZEROFOX_BASE_URL:
https://api.zerofox.com(padrão para a maioria dos locatários).
Verifique as permissões
Para verificar se a conta tem as permissões necessárias:
- Faça login na plataforma ZeroFox.
- Acesse Configurações (⚙️) > Conexões de dados > Feeds de dados da API.
- Se você conseguir ver a seção Feeds de dados da API e gerar tokens, significa que tem as permissões necessárias.
- Se essa opção não aparecer, entre em contato com seu administrador para conceder permissões de acesso à API.
Testar o acesso à API
Teste suas credenciais antes de prosseguir com a integração:
# Replace with your actual credentials ZEROFOX_API_TOKEN="your-personal-access-token" ZEROFOX_BASE_URL="https://api.zerofox.com" # Test API access (example endpoint - adjust based on your data type) curl -v -H "Authorization: Bearer $ZEROFOX_API_TOKEN" \ -H "Accept: application/json" \ "$ZEROFOX_BASE_URL/v1/alerts?limit=1"
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
zerofox-logs-collector-sa. - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect ZeroFox Platform 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 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 à conta de serviço (zerofox-logs-collector-sa) permissões de gravação no bucket do GCS:
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket (por exemplo,
zerofox-platform-logs). - 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,
zerofox-logs-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 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
zerofox-logs-trigger. - Não altere as outras configurações.
- ID do tópico: insira
- 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 da plataforma ZeroFox 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 zerofox-logs-collectorRegiã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):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico do Pub/Sub (
zerofox-logs-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 (
zerofox-logs-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_BUCKETzerofox-platform-logsNome do bucket do GCS GCS_PREFIXzerofox/platformPrefixo para arquivos de registro STATE_KEYzerofox/platform/state.jsonCaminho do arquivo de estado ZEROFOX_BASE_URLhttps://api.zerofox.comURL base da API ZEROFOX_API_TOKENyour-zerofox-personal-access-tokenToken de acesso pessoal LOOKBACK_HOURS24Período de lookback inicial PAGE_SIZE200Registros por página MAX_PAGES20Número máximo de páginas por execução HTTP_TIMEOUT60Tempo limite da solicitação HTTP em segundos HTTP_RETRIES3Número de novas tentativas de HTTP URL_TEMPLATE(opcional) Modelo de URL personalizado com {SINCE},{PAGE_TOKEN},{PAGE_SIZE}Na seção Variáveis e secrets, role a tela para baixo até Solicitações:
- Tempo limite da solicitação: insira
600segundos (10 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 é 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, timedelta import time import urllib.parse # 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', 'zerofox/platform') STATE_KEY = os.environ.get('STATE_KEY', 'zerofox/platform/state.json') ZEROFOX_BASE_URL = os.environ.get('ZEROFOX_BASE_URL', 'https://api.zerofox.com') ZEROFOX_API_TOKEN = os.environ.get('ZEROFOX_API_TOKEN') LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '200')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '60')) HTTP_RETRIES = int(os.environ.get('HTTP_RETRIES', '3')) URL_TEMPLATE = os.environ.get('URL_TEMPLATE', '') def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch ZeroFox Platform logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ZEROFOX_BASE_URL, ZEROFOX_API_TOKEN]): print('Error: Missing required environment variables') return try: # Get GCS bucket 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_since"): try: last_time = parse_datetime(state["last_since"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_since: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) since_iso = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') print(f"Fetching logs since {since_iso}") # Fetch logs records, newest_since = fetch_logs( api_base=ZEROFOX_BASE_URL, api_token=ZEROFOX_API_TOKEN, since=since_iso, page_size=PAGE_SIZE, max_pages=MAX_PAGES, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, since_iso) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest timestamp if newest_since: save_state(bucket, STATE_KEY, newest_since) else: save_state(bucket, STATE_KEY, since_iso) print(f"Successfully processed {len(records)} records") except Exception as e: print(f'Error processing logs: {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_since: str): """Save the last since timestamp to GCS state file.""" try: state = {'last_since': last_since} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_since={last_since}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(api_base: str, api_token: str, since: str, page_size: int, max_pages: int): """ Fetch logs from ZeroFox Platform API with pagination and rate limiting. Args: api_base: API base URL api_token: Personal access token since: ISO timestamp for filtering logs page_size: Number of records per page max_pages: Maximum pages to fetch Returns: Tuple of (records list, newest_since ISO string) """ # Use URL_TEMPLATE if provided, otherwise construct default alerts endpoint if URL_TEMPLATE: base_url = URL_TEMPLATE.replace("{SINCE}", urllib.parse.quote(since)) else: base_url = f"{api_base}/v1/alerts?since={urllib.parse.quote(since)}" headers = { 'Authorization': f'Bearer {api_token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-ZeroFoxCollector/1.0' } records = [] newest_since = since page_num = 0 page_token = "" backoff = 1.0 while page_num < max_pages: page_num += 1 # Construct URL with pagination if URL_TEMPLATE: url = (base_url .replace("{PAGE_TOKEN}", urllib.parse.quote(page_token)) .replace("{PAGE_SIZE}", str(page_size))) else: url = f"{base_url}&limit={page_size}" if page_token: url += f"&page_token={urllib.parse.quote(page_token)}" attempt = 0 while attempt <= HTTP_RETRIES: try: response = http.request('GET', url, headers=headers, timeout=HTTP_TIMEOUT) # Handle rate limiting with exponential backoff 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, 30.0) attempt += 1 continue backoff = 1.0 if response.status != 200: print(f"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return records, newest_since data = json.loads(response.data.decode('utf-8')) # Extract results (try multiple possible keys) page_results = [] for key in ('results', 'data', 'alerts', 'items', 'logs', 'events'): if isinstance(data.get(key), list): page_results = data[key] break if not page_results: print(f"No more results (empty page)") return records, newest_since print(f"Page {page_num}: Retrieved {len(page_results)} events") records.extend(page_results) # Track newest timestamp for event in page_results: try: # Try multiple possible timestamp fields event_time = (event.get('timestamp') or event.get('created_at') or event.get('last_modified') or event.get('event_time') or event.get('log_time') or event.get('updated_at')) if event_time and isinstance(event_time, str): if event_time > newest_since: newest_since = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for next page token next_token = (data.get('next') or data.get('next_token') or data.get('nextPageToken') or data.get('next_page_token')) if isinstance(next_token, dict): next_token = (next_token.get('token') or next_token.get('cursor') or next_token.get('value')) if not next_token: print("No more pages (no next token)") return records, newest_since page_token = str(next_token) break except urllib3.exceptions.HTTPError as e: if attempt < HTTP_RETRIES: print(f"HTTP error (attempt {attempt + 1}/{HTTP_RETRIES}): {e}") time.sleep(1 + attempt) attempt += 1 continue else: print(f"Error fetching logs after {HTTP_RETRIES} retries: {e}") return records, newest_since except Exception as e: print(f"Error fetching logs: {e}") return records, newest_since print(f"Retrieved {len(records)} total records from {page_num} pages") return records, newest_since- Segundo arquivo: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).
Criar o job do Cloud Scheduler
O Cloud Scheduler publica mensagens no tópico do Pub/Sub (zerofox-logs-trigger) 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 zerofox-logs-collector-hourlyRegiã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 Selecione o tópico do Pub/Sub ( zerofox-logs-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 5 minutos */5 * * * *Alto volume e baixa latência A cada 15 minutos */15 * * * *Volume médio A cada hora 0 * * * *Padrão (recomendado) A cada 6 horas 0 */6 * * *Baixo volume, processamento em lote Diário 0 0 * * *Coleta de dados históricos
Testar a integração
- No console do Cloud Scheduler, encontre seu job (
zerofox-logs-collector-hourly). - Clique em Executar à força para acionar o job manualmente.
- Aguarde alguns segundos.
- Acesse Cloud Run > Serviços.
- Clique no nome da função (
zerofox-logs-collector). - Clique na guia Registros.
Verifique se a função foi executada com sucesso. Procure o seguinte:
Fetching logs since YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X events Wrote X records to gs://zerofox-platform-logs/zerofox/platform/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsAcesse Cloud Storage > Buckets.
Clique no nome do bucket (
zerofox-platform-logs).Navegue até a pasta de prefixo (
zerofox/platform/).Verifique se um novo arquivo
.ndjsonfoi criado com o carimbo de data/hora atual.
Se você encontrar erros nos registros:
- HTTP 401: verifique as credenciais da API nas variáveis de ambiente. Verifique se o
ZEROFOX_API_TOKENestá correto e não expirou. - HTTP 403: verifique se a conta do ZeroFox tem as permissões necessárias para alertas e feeds de CTI. Acesse Configurações > Conexões de dados > Feeds de dados da API e verifique as permissões de token.
- HTTP 404: o endpoint
/v1/alertspadrão pode não ser correto para seu locatário. Defina a variável de ambienteURL_TEMPLATEpara corresponder à documentação da API do ZeroFox ou entre em contato com o suporte da ZeroFox. - HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera exponencial.
- Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas na configuração da função do Cloud Run.
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,
ZeroFox Platform Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Plataforma ZeroFox 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.comCopie esse endereço de e-mail para usar na próxima etapa.
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 (
zerofox-platform-logs). - 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.
Configurar um feed no Google SecOps para ingerir registros da plataforma ZeroFox
- 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,
ZeroFox Platform Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Plataforma ZeroFox 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://zerofox-platform-logs/zerofox/platform/Substitua:
zerofox-platform-logs: o nome do bucket do GCS.zerofox/platform: prefixo/caminho da pasta onde os registros são armazenados.
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.
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 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.
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.