Coletar registros de auditoria no nível do grupo do Snyk

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir registros de auditoria no nível do grupo do Snyk no Google Security Operations usando o Google Cloud Storage. Primeiro, o analisador limpa os campos desnecessários dos registros brutos. Em seguida, ele extrai informações relevantes, como detalhes do usuário, tipo de evento e carimbos de data/hora, transformando e mapeando esses dados no esquema UDM do Google SecOps para representação padronizada de registros de segurança.

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 Snyk (administrador do grupo) e um token de API com acesso ao grupo
• Plano Enterprise do Snyk. Os endpoints de registros de auditoria só estão disponíveis nos planos Enterprise.

Coletar pré-requisitos de registros de auditoria no nível do grupo do Snyk (IDs, chaves de API, IDs de organização, tokens)

1. No Snyk, clique no seu avatar > Configurações da conta > Token da API.
2. Clique em Revogar e gerar novamente (ou Gerar) e copie o token.
3. Salve esse token como a variável de ambiente SNYK_API_TOKEN.
4. No Snyk, mude para seu grupo (seletor no canto superior esquerdo).
5. Acesse Configurações do grupo.
6. Copie o <GROUP_ID> do URL: https://app.snyk.io/group/<GROUP_ID>/settings.
   • Ou use a API REST: GET https://api.snyk.io/rest/groups?version=2024-01-04 e escolha o id.
7. Verifique se o usuário do token tem permissão para Ver registros de auditoria (group.audit.read).

Verifique as permissões

Para verificar se a conta tem as permissões necessárias:

1. Faça login no Snyk.
2. Mude para seu grupo (seletor no canto superior esquerdo).
3. Acesse Configurações do grupo.
4. Se a opção Registros de auditoria aparecer na navegação à esquerda, você tem as permissões necessárias.

5. Se você não encontrar essa opção, entre em contato com o administrador para conceder a permissão Ver registros de auditoria (group.audit.read).

   Observação: os endpoints de registros de auditoria exigem um plano do Snyk Enterprise. Se as chamadas de API falharem com erros 403 apesar da configuração correta, verifique seu nível de assinatura.

Testar o acesso à API

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

  # Replace with your actual credentials
  SNYK_API_TOKEN="your-token-here"
  SNYK_GROUP_ID="your-group-id-here"
  SNYK_API_VERSION="2024-01-04"
  
  # Test API access
  curl -v -H "Authorization: token ${SNYK_API_TOKEN}" \
    "https://api.snyk.io/rest/groups/${SNYK_GROUP_ID}/audit_logs/search?version=${SNYK_API_VERSION}&size=10"
  

Criar um bucket do Google Cloud Storage

1. Acesse o Console do Google Cloud.
2. Selecione seu projeto ou crie um novo.
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 exclusivo globalmente, por exemplo, snyk-audit.
   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

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

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 snyk-audit-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Snyk group-level audit 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 do Cloud Functions.
6. Clique em Continuar.
7. 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:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket (por exemplo, snyk-audit).
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, snyk-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
   • 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 snyk-audit-trigger.
   • Não altere as outras configurações.
4. 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 Snyk 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	snyk-audit-collector
   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.

5. 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 o tópico do Pub/Sub (snyk-audit-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 (snyk-audit-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
   GCS_BUCKET	snyk-audit
   GCS_PREFIX	snyk/audit/
   STATE_KEY	snyk/audit/state.json
   SNYK_GROUP_ID	<your_group_id>
   SNYK_API_TOKEN	xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
   SNYK_API_BASE	https://api.snyk.io (opcional)
   SNYK_API_VERSION	2024-01-04
   SIZE	100
   MAX_PAGES	20
   LOOKBACK_SECONDS	3600
   EVENTS	(opcional) group.create,org.user.add
   EXCLUDE_EVENTS	(opcional) api.access

   Observação: o valor SNYK_API_VERSION precisa corresponder à versão recomendada atual da documentação da API de registros de auditoria do Snyk. Consulte a documentação da Snyk periodicamente para garantir que você está usando a versão estável mais recente.

10. Na seção Variáveis e secrets, role a tela para baixo até 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 é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

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

2. 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
   import urllib.parse
   
   # Initialize HTTP client
   http = urllib3.PoolManager()
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch Snyk group-level audit logs 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', 'snyk/audit/')
       state_key = os.environ.get('STATE_KEY', 'snyk/audit/state.json')
   
       # Snyk API configuration
       api_base = os.environ.get('SNYK_API_BASE', 'https://api.snyk.io').rstrip('/')
       group_id = os.environ.get('SNYK_GROUP_ID', '').strip()
       api_token = os.environ.get('SNYK_API_TOKEN', '').strip()
       api_version = os.environ.get('SNYK_API_VERSION', '2024-01-04').strip()
   
       size = int(os.environ.get('SIZE', '100'))
       max_pages = int(os.environ.get('MAX_PAGES', '20'))
       lookback_seconds = int(os.environ.get('LOOKBACK_SECONDS', '3600'))
   
       events_csv = os.environ.get('EVENTS', '').strip()
       exclude_events_csv = os.environ.get('EXCLUDE_EVENTS', '').strip()
   
       if not all([bucket_name, group_id, api_token]):
           print('Error: Missing required environment variables')
           return
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(bucket_name)
   
           # Load state (last cursor)
           state = load_state(bucket, state_key)
           cursor = state.get('cursor')
   
           print(f'Starting log collection with cursor: {cursor}')
   
           # Prepare headers for Snyk REST API
           headers = {
               'Authorization': f'token {api_token}',
               'Accept': 'application/vnd.api+json'
           }
   
           pages = 0
           total = 0
           last_cursor = cursor
   
           # Only for the very first run (no saved cursor), constrain the time window
           first_run_from_iso = None
           if not cursor and lookback_seconds > 0:
               first_run_from_iso = time.strftime(
                   '%Y-%m-%dT%H:%M:%SZ',
                   time.gmtime(time.time() - lookback_seconds)
               )
   
           while pages < max_pages:
               payload = fetch_page(
                   api_base, group_id, headers, api_version, size,
                   cursor, first_run_from_iso, events_csv, exclude_events_csv
               )
   
               # Write payload to GCS
               write_to_gcs(bucket, prefix, payload)
   
               # Extract items count
               data_obj = payload.get('data') or {}
               items = data_obj.get('items') or []
               if isinstance(items, list):
                   total += len(items)
   
               # Parse next cursor
               cursor = parse_next_cursor_from_links(payload.get('links'))
               pages += 1
   
               if not cursor:
                   break
   
               # After first page, disable from-filter
               first_run_from_iso = None
   
           # Save state
           if cursor and cursor != last_cursor:
               save_state(bucket, state_key, {'cursor': cursor})
   
           print(f'Successfully processed {total} events across {pages} pages. Next cursor: {cursor}')
   
       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: {str(e)}')
       return {}
   
   def save_state(bucket, key, state):
       """Save state to GCS."""
       try:
           blob = bucket.blob(key)
           blob.upload_from_string(
               json.dumps(state),
               content_type='application/json'
           )
       except Exception as e:
           print(f'Warning: Could not save state: {str(e)}')
   
   def write_to_gcs(bucket, prefix, payload):
       """Write payload to GCS."""
       ts = time.strftime('%Y/%m/%d/%H%M%S', time.gmtime())
       key = f"{prefix.rstrip('/')}/{ts}-snyk-group-audit.json"
       blob = bucket.blob(key)
       blob.upload_from_string(
           json.dumps(payload, separators=(',', ':')),
           content_type='application/json'
       )
       print(f'Wrote payload to {key}')
   
   def parse_next_cursor_from_links(links):
       """Parse next cursor from links object."""
       if not links:
           return None
       nxt = links.get('next')
       if not nxt:
           return None
       try:
           q = urllib.parse.urlparse(nxt).query
           params = urllib.parse.parse_qs(q)
           cur = params.get('cursor')
           return cur[0] if cur else None
       except Exception:
           return None
   
   def as_list(csv_str):
       """Convert comma-separated string to list."""
       return [x.strip() for x in csv_str.split(',') if x.strip()]
   
   def fetch_page(api_base, group_id, headers, api_version, size, cursor, first_run_from_iso, events_csv, exclude_events_csv):
       """Fetch a single page from Snyk audit logs API."""
       base_path = f'/rest/groups/{group_id}/audit_logs/search'
       params = {
           'version': api_version,
           'size': size,
       }
   
       if cursor:
           params['cursor'] = cursor
       elif first_run_from_iso:
           params['from'] = first_run_from_iso
   
       events = as_list(events_csv)
       exclude_events = as_list(exclude_events_csv)
   
       if events and exclude_events:
           exclude_events = []
   
       if events:
           params['events'] = events
       if exclude_events:
           params['exclude_events'] = exclude_events
   
       url = f"{api_base}{base_path}?{urllib.parse.urlencode(params, doseq=True)}"
   
       response = http.request('GET', url, headers=headers, timeout=60.0)
   
       if response.status == 429 or response.status >= 500:
           retry_after = int(response.headers.get('Retry-After', '1'))
           time.sleep(max(1, retry_after))
           response = http.request('GET', url, headers=headers, timeout=60.0)
   
       if response.status != 200:
           raise Exception(f'API request failed with status {response.status}: {response.data.decode("utf-8")}')
   
       return json.loads(response.data.decode('utf-8'))
   

   • 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 publica 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	snyk-audit-collector-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	Selecione o tópico do Pub/Sub (snyk-audit-trigger).
   Corpo da mensagem	{}

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

1. No console do Cloud Scheduler, encontre seu job (snyk-audit-collector-hourly).
2. Clique em Executar à força para acionar o job manualmente.
3. Aguarde alguns segundos.
4. Acesse Cloud Run > Serviços.
5. Clique no nome da função (snyk-audit-collector).
6. Clique na guia Registros.

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

   Starting log collection with cursor: None
   Page 1: Retrieved X events
   Wrote payload to snyk/audit/YYYY/MM/DD/HHMMSS-snyk-group-audit.json
   Successfully processed X events across Y pages. Next cursor: ...
   

8. Acesse Cloud Storage > Buckets.

9. Clique no nome do bucket (por exemplo, snyk-audit).

10. Navegue até a pasta de prefixo (snyk/audit/).

11. Verifique se um novo arquivo .json foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

• HTTP 401: verifique SNYK_API_TOKEN nas variáveis de ambiente
• HTTP 403: verifique se o usuário do token tem permissão de group.audit.read e se sua assinatura do Snyk é um plano Enterprise.
• 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 (GCS_BUCKET, SNYK_GROUP_ID, SNYK_API_TOKEN).

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

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, Snyk Group Audit Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Registros de auditoria no nível do grupo do Snyk como o Tipo de registro.

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

   [e-mail]
   

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

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket (por exemplo, snyk-audit).
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.

Configurar um feed no Google SecOps para ingerir registros de auditoria no nível do grupo Snyk

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, Snyk Group Audit Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Registros de auditoria no nível do grupo do Snyk como o Tipo de registro.
7. Clique em Próxima.

8. 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://snyk-audit/snyk/audit/
     

     • Substitua:

       • snyk-audit: o nome do bucket do GCS.
       • snyk/audit/: prefixo/caminho da pasta onde os registros são armazenados.
     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: snyk.group_audit

   • Rótulos de ingestão: adicione se quiser.

9. Clique em Próxima.

10. 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
content.url	principal.url	Mapeado diretamente do campo "content.url" no registro bruto.
created	metadata.event_timestamp	Analisado do campo "created" no registro bruto usando o formato ISO8601.
evento	metadata.product_event_type	Mapeado diretamente do campo "event" no registro bruto.
groupId	principal.user.group_identifiers	Mapeado diretamente do campo "groupId" no registro bruto.
orgId	principal.user.attribute.labels.key	Defina como "orgId".
orgId	principal.user.attribute.labels.value	Mapeado diretamente do campo "orgId" no registro bruto.
userId	principal.user.userid	Mapeado diretamente do campo "userId" no registro bruto.
N/A	metadata.event_type	Fixado no código do analisador como "USER_UNCATEGORIZED".
N/A	metadata.log_type	Fixado no código do analisador como "SNYK_SDLC".
N/A	metadata.product_name	Fixado no código do analisador como "SNYK SDLC".
N/A	metadata.vendor_name	Fixado no código do analisador como "SNYK_SDLC".

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