Coletar registros dos serviços de proteção de identidade (IDP) do CrowdStrike

Versão do analisador: 1.0

Compatível com:

Google SecOps SIEM

Neste documento, explicamos como ingerir registros dos serviços de proteção de identidade (IDP) do CrowdStrike no Google Security Operations usando o Google Cloud Storage. A integração usa a API CrowdStrike Unified Alerts para coletar eventos de proteção de identidade e os armazena no formato NDJSON para processamento pelo analisador CS_IDP integrado.

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 console do CrowdStrike Falcon e ao gerenciamento de chaves de API

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 globalmente exclusivo (por exemplo, crowdstrike-idp-logs-bucket)
   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.

Pré-requisitos do CrowdStrike Identity Protection

1. Faça login no console do CrowdStrike Falcon.
2. Acesse Suporte e recursos > Clientes e chaves de API.
3. Clique em Adicionar novo cliente de API.
4. Informe os seguintes detalhes de configuração:
   • Nome do cliente: insira Google SecOps IDP Integration.
   • Descrição: insira API client for Google SecOps integration.
   • Escopos: selecione o escopo Alertas: LEITURA (alerts:read), que inclui os alertas do Identity Protection.
5. Clique em Adicionar.
6. Copie e salve em um local seguro os seguintes detalhes:
   • ID do cliente
   • Chave secreta do cliente (mostrada apenas uma vez)
   • URL de base (exemplos: api.crowdstrike.com para US-1, api.us-2.crowdstrike.com para US-2, api.eu-1.crowdstrike.com para EU-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.

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 crowdstrike-idp-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect CrowdStrike IDP 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: gravar registros no bucket do GCS e gerenciar 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 permissões de gravaçã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 participantes: insira o e-mail da conta de serviço (por exemplo, crowdstrike-idp-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 crowdstrike-idp-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 é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API CrowdStrike Identity Protection 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	crowdstrike-idp-collector
   Região	Selecione a região correspondente ao bucket do GCS (por exemplo, us-central1).
   Ambiente de execução	Selecione Python 3.12 ou 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 crowdstrike-idp-trigger.
   4. Clique em Salvar.

6. Na seção Autenticação:

   1. Selecione Exigir autenticação.
   2. Verifique o Identity and Access Management (IAM).
   Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.

7. Role 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 crowdstrike-idp-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	crowdstrike-idp-logs-bucket
   GCS_PREFIX	crowdstrike-idp/
   STATE_KEY	crowdstrike-idp/state.json
   CROWDSTRIKE_CLIENT_ID	your-client-id
   CROWDSTRIKE_CLIENT_SECRET	your-client-secret
   API_BASE	api.crowdstrike.com (US-1), api.us-2.crowdstrike.com (US-2), api.eu-1.crowdstrike.com (EU-1)
   ALERTS_LIMIT	1000 (opcional, máximo de 10.000 por página)

10. Role 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 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 for criado, o editor de código inline será aberto automaticamente.

Adicionar código da função

1. Digite 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
   from urllib.parse import urlencode
   
   # Initialize HTTP client
   http = urllib3.PoolManager()
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Fetch CrowdStrike Identity Protection alerts (Unified Alerts API)
       and store RAW JSON (NDJSON) to GCS for the CS_IDP parser.
       No transformation is performed.
       """
   
       # Get environment variables
       bucket_name = os.environ.get('GCS_BUCKET')
       prefix = os.environ.get('GCS_PREFIX', 'crowdstrike-idp/')
       state_key = os.environ.get('STATE_KEY', 'crowdstrike-idp/state.json')
       client_id = os.environ.get('CROWDSTRIKE_CLIENT_ID')
       client_secret = os.environ.get('CROWDSTRIKE_CLIENT_SECRET')
       api_base = os.environ.get('API_BASE')
   
       if not all([bucket_name, client_id, client_secret, api_base]):
           print('Error: Missing required environment variables')
           return
   
       try:
           bucket = storage_client.bucket(bucket_name)
   
           # Get OAuth token
           token = get_token(client_id, client_secret, api_base)
   
           # Load last processed timestamp
           last_ts = get_last_timestamp(bucket, state_key)
   
           # FQL filter for Identity Protection alerts only, newer than checkpoint
           fql_filter = f"product:'idp' + updated_timestamp:>'{last_ts}'"
           sort = 'updated_timestamp.asc'
   
           # Step 1: Get list of alert IDs
           all_ids = []
           per_page = int(os.environ.get('ALERTS_LIMIT', '1000'))
           offset = 0
   
           while True:
               page_ids = query_alert_ids(api_base, token, fql_filter, sort, per_page, offset)
               if not page_ids:
                   break
               all_ids.extend(page_ids)
               if len(page_ids) < per_page:
                   break
               offset += per_page
   
           if not all_ids:
               print('No new Identity Protection alerts.')
               return
   
           # Step 2: Get alert details in batches (max 1000 IDs per request)
           details = []
           max_batch = 1000
           for i in range(0, len(all_ids), max_batch):
               batch = all_ids[i:i + max_batch]
               details.extend(fetch_alert_details(api_base, token, batch))
   
           if details:
               # Sort by updated_timestamp
               details.sort(key=lambda d: d.get('updated_timestamp', d.get('created_timestamp', '')))
               latest = details[-1].get('updated_timestamp') or details[-1].get('created_timestamp')
   
               # Write to GCS
               timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
               blob_name = f"{prefix}cs_idp_{timestamp}.json"
               blob = bucket.blob(blob_name)
   
               # NDJSON format
               log_lines = '\n'.join([json.dumps(d, separators=(',', ':')) for d in details])
               blob.upload_from_string(log_lines, content_type='application/x-ndjson')
   
               print(f'Wrote {len(details)} alerts to {blob_name}')
   
               # Update state
               update_state(bucket, state_key, latest)
   
       except Exception as e:
           print(f'Error processing alerts: {str(e)}')
           raise
   
   def get_token(client_id, client_secret, api_base):
       """Get OAuth2 token from CrowdStrike API"""
       url = f"https://{api_base}/oauth2/token"
       data = f"client_id={client_id}&client_secret={client_secret}&grant_type=client_credentials"
       headers = {'Content-Type': 'application/x-www-form-urlencoded'}
   
       r = http.request('POST', url, body=data, headers=headers)
       if r.status != 200:
           raise Exception(f'Auth failed: {r.status} {r.data}')
   
       return json.loads(r.data.decode('utf-8'))['access_token']
   
   def query_alert_ids(api_base, token, fql_filter, sort, limit, offset):
       """Query alert IDs using filters"""
       url = f"https://{api_base}/alerts/queries/alerts/v2"
       params = {
           'filter': fql_filter,
           'sort': sort,
           'limit': str(limit),
           'offset': str(offset)
       }
       qs = urlencode(params)
   
       r = http.request('GET', f"{url}?{qs}", headers={'Authorization': f'Bearer {token}'})
       if r.status != 200:
           raise Exception(f'Query alerts failed: {r.status} {r.data}')
   
       resp = json.loads(r.data.decode('utf-8'))
       return resp.get('resources', [])
   
   def fetch_alert_details(api_base, token, composite_ids):
       """Fetch detailed alert data by composite IDs"""
       url = f"https://{api_base}/alerts/entities/alerts/v2"
       body = {'composite_ids': composite_ids}
       headers = {
           'Authorization': f'Bearer {token}',
           'Content-Type': 'application/json'
       }
   
       r = http.request('POST', url, body=json.dumps(body).encode('utf-8'), headers=headers)
       if r.status != 200:
           raise Exception(f'Fetch alert details failed: {r.status} {r.data}')
   
       resp = json.loads(r.data.decode('utf-8'))
       return resp.get('resources', [])
   
   def get_last_timestamp(bucket, key, default='2023-01-01T00:00:00Z'):
       """Get last processed timestamp from GCS state file"""
       try:
           blob = bucket.blob(key)
           if blob.exists():
               state_data = blob.download_as_text()
               state = json.loads(state_data)
               return state.get('last_timestamp', default)
       except Exception as e:
           print(f'Warning: Could not load state: {str(e)}')
       return default
   
   def update_state(bucket, key, ts):
       """Update last processed timestamp in GCS state file"""
       state = {
           'last_timestamp': ts,
           'updated': datetime.now(timezone.utc).isoformat()
       }
       blob = bucket.blob(key)
       blob.upload_from_string(json.dumps(state), content_type='application/json')
   

   • 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).

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	crowdstrike-idp-collector-15m
   Região	Selecione a mesma região da função do Cloud Run
   Frequência	*/15 * * * * (a cada 15 minutos)
   Fuso horário	Selecione o fuso horário (UTC recomendado)
   Tipo de destino	Pub/Sub
   Tópico	Selecione o tópico crowdstrike-idp-trigger.
   Corpo da mensagem	{} (objeto JSON vazio)

4. Clique em Criar.

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 > crowdstrike-idp-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 bucket do GCS. É necessário conceder a essa conta de serviço acesso ao bucket.

Encontrar o e-mail da conta de serviço

1. Acesse Configurações > Gerenciamento de dados > Feeds. Se você estiver em uma plataforma somente SIEM, acesse Configurações > 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, CrowdStrike Identity Protection Services logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Serviços de proteção de identidade do Crowdstrike como o Tipo de registro.

7. Clique em Get Service Account. Um e-mail exclusivo da conta de serviço será 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 Leitor de objetos do Storage 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: 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 de Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar um feed no Google SecOps para ingerir registros dos serviços de proteção de identidade do CrowdStrike

1. Acesse Configurações > Gerenciamento de dados > Feeds. Se você estiver em uma plataforma somente SIEM, acesse Configurações > 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, CrowdStrike Identity Protection Services logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Serviços de proteção de identidade do Crowdstrike 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://crowdstrike-idp-logs-bucket/crowdstrike-idp/
     

     • Substitua:

       • crowdstrike-idp-logs-bucket: o nome do bucket do GCS.
       • crowdstrike-idp/: caminho do prefixo/pasta em que 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 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 de Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo.

   • Idade máxima do arquivo: inclua arquivos modificados nos últimos 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 desse feed.

9. Clique em Próxima.

10. Revise a configuração do novo feed na tela Finalizar e clique em Enviar.

Registro de alterações

Confira o registro de alterações deste analisador

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