Coletar registros de problemas do Snyk Group

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 problemas do Snyk Group no Google Security Operations usando o Google Cloud Storage. O Snyk é uma plataforma de segurança para desenvolvedores que ajuda as organizações a encontrar e corrigir vulnerabilidades em dependências de código aberto, imagens de contêiner, configurações de infraestrutura como código e código de aplicativo. Os problemas do grupo do Snyk oferecem visibilidade das vulnerabilidades de segurança e dos problemas de licença em todos os projetos de um grupo do Snyk.

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 funções do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
• Acesso privilegiado ao grupo do Snyk (token da API com acesso de leitura e ID do grupo)
• Função de administrador do grupo Snyk atribuída ao usuário com o token da API. O usuário precisa poder ver os registros de auditoria e os problemas do grupo.

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

6. Clique em Criar.

Coletar o ID do grupo e o token da API do Snyk

Receber token da API do Snyk

1. Faça login na interface do Snyk em https://app.snyk.io.
2. Acesse Configurações da conta > Token da API.
3. Clique em Gerar para gerar o token da API.

4. Copie e salve o token em um local seguro para usar mais tarde como SNYK_TOKEN.

   Observação: esse token será usado pela função do Cloud Run para autenticar a API Snyk.

Receber ID do grupo do Snyk

1. Na interface do Snyk, mude para seu grupo.
2. Acesse Configurações do grupo.
3. Copie e salve o ID do grupo do URL (https://app.snyk.io/group/<GROUP_ID>/...) para usar depois como GROUP_ID.

Atribuir a função de administrador do grupo

1. Na interface do Snyk, acesse Configurações do grupo > Membros.
2. Localize o usuário associado ao token da API.

3. Atribua a função Administrador do grupo ao usuário.

   Observação: o usuário precisa ter permissão para acessar os registros de auditoria e os problemas do grupo.

Endpoint da API Note

• O endpoint de base da API REST varia de acordo com a região. Identifique sua região do Snyk e anote o URL base REST correspondente:

  Região	URL base do REST
  SNYK-US-01	https://api.snyk.io/rest
  SNYK-US-02	https://api.us.snyk.io/rest
  SNYK-EU-01	https://api.eu.snyk.io/rest
  SNYK-AU-01	https://api.au.snyk.io/rest

  Você vai usar esse URL base REST como API_BASE na configuração da função do Cloud Run. O código da função cria URLs de endpoint completos anexando caminhos como /groups/{group_id}/audit_logs/search a esse URL de base.

  Observação: os endpoints de registros de auditoria do Snyk estão disponíveis apenas para planos Enterprise. Verifique se o seu plano inclui acesso aos registros de auditoria de grupo e se sua conta tem os limites de retenção adequados configurados.

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.

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-logs-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Snyk Group logs.
4. Clique em Criar e continuar.
5. Na seção Conceda a essa conta de serviço acesso ao projeto:
   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.
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-logs-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-logs-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 do Snyk Group 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-group-logs-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 snyk-logs-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-logs-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-group-logs
   GCS_PREFIX	snyk/group/
   STATE_KEY	snyk/group/state.json
   SNYK_TOKEN	your-snyk-api-token
   GROUP_ID	your-group-uuid
   API_BASE	https://api.snyk.io/rest
   SNYK_AUDIT_API_VERSION	2024-10-15
   SNYK_ISSUES_API_VERSION	2024-10-15
   AUDIT_PAGE_SIZE	100
   ISSUES_PAGE_LIMIT	100
   MAX_PAGES	20
   LOOKBACK_SECONDS	3600

   Observação: defina API_BASE como o URL base REST correto para sua região do Snyk. Para SNYK-US-01, use https://api.snyk.io/rest. Para SNYK-EU-01, use https://api.eu.snyk.io/rest. Para SNYK-US-02, use https://api.us.snyk.io/rest. Para SNYK-AU-01, use https://api.au.snyk.io/rest.

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.
    • Clique em Concluído.

12. Role a tela para baixo até Ambiente de execução:

    • Selecione Padrão (recomendado).

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

14. Clique em Criar.

15. Aguarde a criação do serviço (1 a 2 minutos).

16. 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 time
         import urllib.parse
         from urllib.request import Request, urlopen
         from urllib.parse import urlparse, parse_qs
         from urllib.error import HTTPError
     
         # Initialize Storage client
         storage_client = storage.Client()
     
         @functions_framework.cloud_event
         def main(cloud_event):
             """
             Cloud Run function triggered by Pub/Sub to fetch logs from Snyk Group API 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/group/').strip()
             state_key = os.environ.get('STATE_KEY', 'snyk/group/state.json').strip()
     
             # Snyk API credentials
             api_base = os.environ.get('API_BASE', 'https://api.snyk.io/rest').rstrip('/')
             snyk_token = os.environ.get('SNYK_TOKEN').strip()
             group_id = os.environ.get('GROUP_ID').strip()
     
             # Page sizes & limits
             audit_size = int(os.environ.get('AUDIT_PAGE_SIZE', '100'))
             issues_limit = int(os.environ.get('ISSUES_PAGE_LIMIT', '100'))
             max_pages = int(os.environ.get('MAX_PAGES', '20'))
     
             # API versions
             audit_api_version = os.environ.get('SNYK_AUDIT_API_VERSION', '2024-10-15').strip()
             issues_api_version = os.environ.get('SNYK_ISSUES_API_VERSION', '2024-10-15').strip()
     
             # First-run lookback
             lookback_seconds = int(os.environ.get('LOOKBACK_SECONDS', '3600'))
     
             if not all([bucket_name, snyk_token, group_id]):
                 print('Error: Missing required environment variables')
                 return
     
             try:
                 # Get GCS bucket
                 bucket = storage_client.bucket(bucket_name)
     
                 # Load state
                 state = load_state(bucket, state_key)
     
                 print('Starting Snyk Group logs collection')
     
                 # Pull audit logs
                 audit_res = pull_audit_logs(
                     bucket, prefix, state, api_base, snyk_token, group_id,
                     audit_api_version, audit_size, max_pages, lookback_seconds
                 )
                 print(f"Audit logs: {audit_res}")
     
                 # Pull issues
                 issues_res = pull_issues(
                     bucket, prefix, state, api_base, snyk_token, group_id,
                     issues_api_version, issues_limit, max_pages
                 )
                 print(f"Issues: {issues_res}")
     
                 # Save state
                 save_state(bucket, state_key, state)
     
                 print('Successfully completed Snyk Group logs collection')
     
             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, separators=(',', ':')),
                     content_type='application/json'
                 )
             except Exception as e:
                 print(f'Warning: Could not save state: {str(e)}')
     
         def _iso(ts):
             """Convert timestamp to ISO format."""
             return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(ts))
     
         def _http_get(url, headers):
             """Make HTTP GET request with retry logic."""
             req = Request(url, method='GET', headers=headers)
             try:
                 with urlopen(req, timeout=60) as r:
                     return json.loads(r.read().decode('utf-8'))
             except HTTPError as e:
                 if e.code in (429, 500, 502, 503, 504):
                     delay = int(e.headers.get('Retry-After', '1'))
                     time.sleep(max(1, delay))
                     with urlopen(req, timeout=60) as r2:
                         return json.loads(r2.read().decode('utf-8'))
                 raise
     
         def _write_page(bucket, prefix, kind, payload):
             """Write page to GCS."""
             ts = time.gmtime()
             key = f"{prefix.rstrip('/')}/{time.strftime('%Y/%m/%d/%H%M%S', ts)}-snyk-{kind}.json"
             blob = bucket.blob(key)
             blob.upload_from_string(
                 json.dumps(payload, separators=(',', ':')),
                 content_type='application/json'
             )
             return key
     
         def _next_href(links):
             """Extract next href from links."""
             if not links:
                 return None
             nxt = links.get('next')
             if not nxt:
                 return None
             if isinstance(nxt, str):
                 return nxt
             if isinstance(nxt, dict):
                 return nxt.get('href')
             return None
     
         def pull_audit_logs(bucket, prefix, state, api_base, snyk_token, group_id,
                             audit_api_version, audit_size, max_pages, lookback_seconds):
             """Pull audit logs from Snyk Group API."""
             headers = {
                 'Authorization': f'token {snyk_token}',
                 'Accept': 'application/vnd.api+json',
             }
     
             cursor = state.get('audit_cursor')
             pages = 0
             total = 0
             base = f"{api_base}/groups/{group_id}/audit_logs/search"
             params = {
                 'version': audit_api_version,
                 'size': audit_size
             }
     
             if cursor:
                 params['cursor'] = cursor
             else:
                 now = time.time()
                 params['from'] = _iso(now - lookback_seconds)
                 params['to'] = _iso(now)
     
             while pages < max_pages:
                 url = f"{base}?{urllib.parse.urlencode(params, doseq=True)}"
                 payload = _http_get(url, headers)
                 _write_page(bucket, prefix, 'audit', payload)
     
                 data_items = (payload.get('data') or {}).get('items') or []
                 if isinstance(data_items, list):
                     total += len(data_items)
     
                 nxt = _next_href(payload.get('links'))
                 if not nxt:
                     break
     
                 q = parse_qs(urlparse(nxt).query)
                 cur = (q.get('cursor') or [None])[0]
                 if not cur:
                     break
     
                 params = {
                     'version': audit_api_version,
                     'size': audit_size,
                     'cursor': cur
                 }
                 state['audit_cursor'] = cur
                 pages += 1
     
             return {
                 'pages': pages + 1 if total else pages,
                 'items': total,
                 'cursor': state.get('audit_cursor')
             }
     
         def pull_issues(bucket, prefix, state, api_base, snyk_token, group_id,
                         issues_api_version, issues_limit, max_pages):
             """Pull issues from Snyk Group API."""
             headers = {
                 'Authorization': f'token {snyk_token}',
                 'Accept': 'application/vnd.api+json',
             }
     
             cursor = state.get('issues_cursor')
             pages = 0
             total = 0
             base = f"{api_base}/groups/{group_id}/issues"
             params = {
                 'version': issues_api_version,
                 'limit': issues_limit
             }
     
             if cursor:
                 params['starting_after'] = cursor
     
             while pages < max_pages:
                 url = f"{base}?{urllib.parse.urlencode(params, doseq=True)}"
                 payload = _http_get(url, headers)
                 _write_page(bucket, prefix, 'issues', payload)
     
                 data_items = payload.get('data') or []
                 if isinstance(data_items, list):
                     total += len(data_items)
     
                 nxt = _next_href(payload.get('links'))
                 if not nxt:
                     break
     
                 q = parse_qs(urlparse(nxt).query)
                 cur = (q.get('starting_after') or [None])[0]
                 if not cur:
                     break
     
                 params = {
                     'version': issues_api_version,
                     'limit': issues_limit,
                     'starting_after': cur
                 }
                 state['issues_cursor'] = cur
                 pages += 1
     
             return {
                 'pages': pages + 1 if total else pages,
                 'items': total,
                 'cursor': state.get('issues_cursor')
             }
         ```
     

   • Segundo arquivo: requirements.txt:

     functions-framework==3.*
     google-cloud-storage==2.*
     

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-group-logs-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 snyk-logs-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 (recomendado)
  A cada 6 horas	0 */6 * * *	Baixo volume, processamento em lote
  Diário	0 0 * * *	Coleta de dados históricos

Testar o job do programador

1. No console do Cloud Scheduler, encontre seu job.
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > snyk-group-logs-collector > Registros.
4. Verifique se a função foi executada com sucesso.
5. Verifique o bucket do GCS para confirmar se os registros foram gravados.

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/Issues).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Registros de auditoria/problemas 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.
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 do Snyk Group

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/Issues).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Registros de auditoria/problemas 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-group-logs/snyk/group/
     

     • Substitua:

       • snyk-group-logs: o nome do bucket do GCS.
       • snyk/group/: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).

     • Exemplos:

       • Bucket raiz: gs://company-logs/
       • Com prefixo: gs://company-logs/snyk-logs/
       • Com subpasta: gs://company-logs/snyk/group/
   Observação: sempre inclua a barra (/) no final do URI.
   • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:
     • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
     • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.
     • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.
   Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.
   • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
   • Namespace do recurso: o namespace do recurso (por exemplo, snyk.group).
   • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

9. Clique em Próxima.

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