Coletar registros do SpyCloud

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros do SpyCloud no Google Security Operations usando o Google Cloud Storage V2.

A SpyCloud é uma plataforma de prevenção contra invasão de contas que oferece inteligência sobre violações e credenciais roubadas. Ele fornece registros de violação, alertas de lista de observação e relatórios de credenciais comprometidas por uma API REST.

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 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
• Uma conta da SpyCloud com acesso à API e uma chave de API válida

Criar um bucket do Google Cloud Storage

1. Acesse o Console do Google 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, spycloud-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 SpyCloud

Receber chave de API

1. Faça login no portal da SpyCloud como administrador.
2. Acesse Configurações > API.
3. Copie e salve a chave de API em um local seguro.

Verifique as permissões

Para verificar se a chave de API tem o acesso necessário:

1. Faça login no portal da SpyCloud.
2. Acesse Configurações > API.
3. Verifique se a chave de API está ativa e tem acesso aos endpoints necessários (dados de violação, lista de observação, bússola).
4. Se o acesso for restrito, entre em contato com o administrador do SpyCloud.

Testar o acesso à API

• Teste suas credenciais antes de prosseguir com a integração:

  # Replace with your actual API key
  API_KEY="your-api-key"
  
  # Test API access - fetch watchlist data
  curl -v -H "X-API-Key: ${API_KEY}" \
    "https://api.spycloud.io/enterprise-v2/breach/data/watchlist?since=2024-01-01&until=2024-01-02"
  

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

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 spycloud-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect SpyCloud 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 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:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket (por exemplo, spycloud-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, [e-mail]).
   • Atribuir papéis: selecione Administrador de objetos do Storage.
6. 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.

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 spycloud-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 registros da API SpyCloud e gravá-los no GCS.

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	spycloud-collector
   Regiã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.

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 spycloud-trigger.
   4. Clique em Salvar.

6. 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.

7. Role a tela para baixo e abra Contêineres, rede, segurança.

8. Acesse a guia Segurança:

   • Conta de serviço: selecione a conta de serviço spycloud-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	spycloud-logs	Nome do bucket do GCS
   GCS_PREFIX	spycloud	Prefixo para arquivos de registro
   STATE_KEY	spycloud/state.json	Caminho do arquivo de estado
   API_KEY	your-api-key	Chave de API do SpyCloud
   API_BASE	https://api.spycloud.io	URL de base da API
   MAX_RECORDS	10000	Máximo de registros por execução
   LOOKBACK_DAYS	7	Período inicial de lookback em dias
   STREAMS	watchlist,catalog	Fluxos de dados separados por vírgulas

10. Role a tela para baixo na guia Variáveis e secrets até Solicitações:

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

11. Acesse a guia Configurações em Contêineres:

    • 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 código da função

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

2. 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
     
     # 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', 'spycloud')
     STATE_KEY = os.environ.get('STATE_KEY', 'spycloud/state.json')
     API_KEY = os.environ.get('API_KEY', '')
     API_BASE = os.environ.get('API_BASE', 'https://api.spycloud.io').rstrip('/')
     MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
     LOOKBACK_DAYS = int(os.environ.get('LOOKBACK_DAYS', '7'))
     STREAMS = [s.strip() for s in os.environ.get('STREAMS', 'watchlist').split(',') if s.strip()]
     
     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 SpyCloud logs and write to GCS.
     
       Args:
         cloud_event: CloudEvent object containing Pub/Sub message
       """
     
       if not all([GCS_BUCKET, API_KEY]):
         print('Error: Missing required environment variables')
         return
     
       try:
         bucket = storage_client.bucket(GCS_BUCKET)
     
         # Load state
         state = load_state(bucket, STATE_KEY)
     
         now = datetime.now(timezone.utc)
     
         # Determine date range
         if isinstance(state, dict) and state.get("last_date"):
           since_date = state["last_date"]
         else:
           since_date = (now - timedelta(days=LOOKBACK_DAYS)).strftime('%Y-%m-%d')
     
         until_date = now.strftime('%Y-%m-%d')
     
         print(f"Fetching data from {since_date} to {until_date}")
     
         report = {}
     
         if 'watchlist' in STREAMS:
           print("Fetching watchlist breach data...")
           count = pull_watchlist(bucket, since_date, until_date)
           report['watchlist_records'] = count
     
         if 'catalog' in STREAMS:
           print("Fetching breach catalog...")
           count = pull_catalog(bucket, since_date, until_date)
           report['catalog_records'] = count
     
         if 'compass' in STREAMS:
           print("Fetching compass data...")
           count = pull_compass(bucket, since_date, until_date)
           report['compass_records'] = count
     
         # Update state
         save_state(bucket, STATE_KEY, until_date)
         print(f"Successfully processed: {json.dumps(report)}")
     
       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_date: str):
       """Save the last query date to GCS state file."""
       try:
         state = {'last_date': last_date, 'last_event_time': datetime.now(timezone.utc).isoformat()}
         blob = bucket.blob(key)
         blob.upload_from_string(
           json.dumps(state, indent=2),
           content_type='application/json'
         )
         print(f"Saved state: last_date={last_date}")
       except Exception as e:
         print(f"Warning: Could not save state: {e}")
     
     def api_get(endpoint: str, params: dict = None):
       """Make authenticated GET request to SpyCloud API with rate limiting."""
       url = f"{API_BASE}{endpoint}"
       if params:
         query = '&'.join([f"{k}={v}" for k, v in params.items()])
         url = f"{url}?{query}"
     
       headers = {
         'X-API-Key': API_KEY,
         'Accept': 'application/json',
         'User-Agent': 'GoogleSecOps-SpyCloudCollector/1.0'
       }
     
       backoff = 1.0
       max_retries = 3
     
       for attempt in range(max_retries):
         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, 30.0)
           continue
     
         if response.status != 200:
           print(f"HTTP Error: {response.status} - {response.data.decode('utf-8')}")
           return None
     
         return json.loads(response.data.decode('utf-8'))
     
       print(f"Failed after {max_retries} retries due to rate limiting")
       return None
     
     def write_ndjson(bucket, prefix: str, stream_name: str, records: list):
       """Write records to GCS as NDJSON."""
       if not records:
         return 0
     
       now = datetime.now(timezone.utc)
       timestamp = now.strftime('%Y%m%d_%H%M%S')
       object_key = f"{GCS_PREFIX}/{stream_name}/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}")
       return len(records)
     
     def pull_watchlist(bucket, since_date: str, until_date: str):
       """Fetch watchlist breach data."""
       cursor = None
       all_records = []
     
       while True:
         params = {'since': since_date, 'until': until_date}
         if cursor:
           params['cursor'] = cursor
     
         data = api_get('/enterprise-v2/breach/data/watchlist', params)
         if not data:
           break
     
         results = data.get('results', [])
         if not results:
           break
     
         all_records.extend(results)
     
         if len(all_records) >= MAX_RECORDS:
           print(f"Reached max_records limit ({MAX_RECORDS})")
           break
     
         cursor = data.get('cursor')
         if not cursor:
           break
     
       return write_ndjson(bucket, GCS_PREFIX, 'watchlist', all_records)
     
     def pull_catalog(bucket, since_date: str, until_date: str):
       """Fetch breach catalog."""
       params = {'since': since_date, 'until': until_date}
       data = api_get('/enterprise-v2/breach/catalog', params)
       if not data:
         return 0
     
       results = data.get('results', [])
       return write_ndjson(bucket, GCS_PREFIX, 'catalog', results)
     
     def pull_compass(bucket, since_date: str, until_date: str):
       """Fetch compass findings."""
       cursor = None
       all_records = []
     
       while True:
         params = {'since': since_date, 'until': until_date}
         if cursor:
           params['cursor'] = cursor
     
         data = api_get('/enterprise-v2/compass/data', params)
         if not data:
           break
     
         results = data.get('results', [])
         if not results:
           break
     
         all_records.extend(results)
     
         if len(all_records) >= MAX_RECORDS:
           print(f"Reached max_records limit ({MAX_RECORDS})")
           break
     
         cursor = data.get('cursor')
         if not cursor:
           break
     
       return write_ndjson(bucket, GCS_PREFIX, 'compass', all_records)
     

   • Segundo arquivo: requirements.txt:

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

3. Clique em Implantar para salvar e implantar a função.

4. 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 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	spycloud-collector-daily
   Região	Selecione a mesma região da função do Cloud Run
   Frequência	0 0 * * * (diariamente à meia-noite)
   Fuso horário	Selecionar fuso horário (UTC recomendado)
   Tipo de destino	Pub/Sub
   Tópico	Selecione o tópico spycloud-trigger.
   Corpo da mensagem	{} (objeto JSON vazio)

4. 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
A cada 6 horas	0 */6 * * *	Baixo volume, processamento em lote
Diário	0 0 * * *	Coleta de dados históricos (recomendada para dados de violação)

Testar a integração

1. No console do Cloud Scheduler, encontre seu job (spycloud-collector-daily).
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > spycloud-collector > Registros.

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

   Fetching data from YYYY-MM-DD to YYYY-MM-DD
   Fetching watchlist breach data...
   Wrote X records to gs://spycloud-logs/spycloud/watchlist/logs_YYYYMMDD_HHMMSS.ndjson
   Successfully processed: {"watchlist_records": X, "catalog_records": Y}
   

5. Verifique o bucket do GCS (spycloud-logs) para confirmar se os registros foram gravados.

Se você encontrar erros nos registros:

• HTTP 401: verifique a chave de API nas variáveis de ambiente
• HTTP 403: verifique se a chave de API tem acesso aos endpoints necessários
• HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera.
• 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 do SpyCloud

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, SpyCloud Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione SpyCloud como o Tipo de registro.

7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:

   [e-mail]
   

8. Copie o endereço de e-mail. Você vai usá-la 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.

9. Clique em Próxima.

10. 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://spycloud-logs/spycloud/
      

      • Substitua:
        • spycloud-logs: o nome do bucket do GCS.
        • spycloud: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
    Observação: sempre inclua a barra invertida (/) 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.

      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência.

        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: 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.

11. Clique em Próxima.

12. 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.

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket (spycloud-logs).
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
   • Atribuir papéis: selecione Leitor de objetos do Storage.

6. 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.

Tabela de mapeamento de UDM

Campo de registro	Mapeamento do UDM	Lógica
av_softwares, assets.av_softwares, assets.country, assets.country_code, assets.display_resolution, assets.email, assets.full_name, assets.infected_machine_id, assets.infected_path, assets.infected_time, assets.ip_addresses, assets.keyboard_languages, assets.password, assets.target_url, assets.username, assets.user_browser, assets.user_hostname, assets.user_os, assets.user_sys_registered_owner	additional.fields	Unidos de rótulos criados com base nesses campos e em av_software_list
senha	extensions.auth.auth_details	Valor copiado diretamente
quando	metadata.event_timestamp	Analisado como carimbo de data/hora ISO8601
	metadata.event_type	Definido com base nas condições: NETWORK_CONNECTION se has_principal, has_target, has_network; USER_UNCATEGORIZED se has_principal e has_principal_userid; STATUS_UPDATE se has_principal e não has_principal_ip; caso contrário, GENERIC_EVENT; ou USER_UNCATEGORIZED se user_hostname, ip ou infected_machine_id estiverem presentes
infected_time	metadata.ingested_timestamp	Analisado como carimbo de data/hora com formatos aaaa-MM-ddTHH:mm:ssZ, RFC 3339, ISO8601
log_id, assets.log_id, uuid	metadata.product_log_id	Valor de log_id se não estiver vazio, caso contrário, assets.log_id ou uuid
user_os	network.http.parsed_user_agent	Convertido para user agent analisado
user_os	network.http.user_agent	Valor copiado diretamente
cookie_domain	principal.administrative_domain	Valor copiado diretamente
país	principal.asset.location.country_or_region	Valor copiado diretamente
infected_machine_id	principal.asset_id	Concatenado como "id: " + infected_machine_id
infected_path	principal.file.full_path	Valor copiado diretamente
user_hostname, domain	principal.hostname	Definido como user_hostname se não estiver vazio e como domínio se não estiver vazio.
ip, ip_addresses	principal.ip	Unido das matrizes "ip" e "ip_addresses"
country_code	principal.resource.attribute.labels	Unido de "country_code_label" criado com base em "country_code"
ID	principal.resource.id	Valor copiado diretamente (convertido em string)
página inicial	principal.url	Valor copiado diretamente
e-mail	principal.user.email_addresses	Valor copiado diretamente
full_name	principal.user.user_display_name	Valor copiado diretamente
user_sys_registered_owner, email_username	principal.user.userid	Definido como user_sys_registered_owner e depois como email_username se não estiver vazio
confiança	security_result.confidence_details	Convertido em string
descrição	security_result.description	Valor copiado diretamente
cookie_expiration, cookie_name, cookie_subdomain, cookie_value, day, document_id, locality_zone, source_id, spycloud_publishdate, spycloud_publish_date, user_browser, infected_time, timezone, password_type, password_plaintext, email_domain, api_token, account_status, breach_category, breach_main_category, consumer_category, malware_family, num_records, premium_flag, sensitive_source, short_title, site_description, title, tlp, type, display_resolution, keyboard_languages	security_result.detection_fields	Unidos de rótulos criados com base nesses campos
gravidade,	security_result.severity	Defina como LOW se "2", INFORMATIONAL se "5", HIGH se "20", CRITICAL se "25" ou "26"
gravidade,	security_result.severity_details	Valor copiado diretamente
target_subdomain	target.administrative_domain	Valor copiado diretamente
target_domain	target.asset.hostname	Valor copiado diretamente
target_domain	target.hostname	Valor copiado diretamente
target_url	target.url	Valor copiado diretamente
nome de usuário	target.user.userid	Valor copiado diretamente
	metadata.product_name	Defina como "SPYCLOUD"
	metadata.vendor_name	Defina como "SpyCloud".
spycloud_publish_date", "infected_time", and "timezone	security_result.detection_fields	Mapeado do registro de mudanças

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.