Coletar registros de fluxos de eventos do CrowdStrike Falcon

Versão do analisador:12.0

Compatível com:

Este documento explica como ingerir registros de fluxos de eventos do CrowdStrike Falcon no Google Security Operations usando o Google Cloud Storage V2. O CrowdStrike Falcon Event Streams oferece uma API de streaming em tempo real que fornece dados de ocorrência de segurança da plataforma Falcon, incluindo eventos de detecção, eventos de auditoria, atividade de autenticação e atualizações de incidentes. A API Event Streams usa uma conexão HTTP persistente com o endpoint /sensors/entities/datafeed/v2 para enviar eventos quase em tempo real, oferecendo suporte à retomada baseada em deslocamento para uma entrega confiável.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Instância do Google SecOps.
  • Um projeto do GCP com a API 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 com permissões para criar clientes de API.

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, cs-stream-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.

Configurar o acesso à API do CrowdStrike Falcon

Criar cliente de API

  1. Faça login no console do CrowdStrike Falcon.
  2. No menu de navegação, acesse Suporte e recursos > Recursos e ferramentas > Clientes e chaves de API.
  3. Clique em Criar cliente de API.
  4. Informe os seguintes detalhes de configuração:
    • Nome do cliente: insira um nome descritivo (por exemplo, Google SecOps Event Streams Integration).
    • Descrição (opcional): insira uma descrição (por exemplo, API client for streaming events to GCS).
  5. Na seção Escopos da API, selecione a seguinte permissão:
    • Fluxos de eventos: selecione Ler (obrigatório para consumir a API de streaming).
  6. Clique em Criar.

Registrar credenciais de API

Depois de criar o cliente de API, você vai receber as seguintes credenciais:

  • ID do cliente: uma string hexadecimal de 32 caracteres em letras minúsculas.
  • Chave secreta do cliente: uma string alfanumérica de 40 caracteres.
  • URL base: o nome de domínio totalmente qualificado da API CrowdStrike (por exemplo, api.crowdstrike.com ou api.us-2.crowdstrike.com).

Importante: copie e salve a chave secreta do cliente imediatamente. Ele é exibido apenas uma vez e não pode ser recuperado depois.

Endpoints regionais

A CrowdStrike opera em várias regiões com diferentes endpoints de API:

Região URL de base
US-1 api.crowdstrike.com
US-2 api.us-2.crowdstrike.com
EU-1 api.eu-1.crowdstrike.com
US-GOV-1 api.laggar.gcw.crowdstrike.com

Importante: use o URL de base que corresponde à região de login do console do CrowdStrike Falcon.

Verifique as permissões

Para verificar se o cliente da API tem as permissões necessárias:

  1. Faça login no console do CrowdStrike Falcon.
  2. Acesse Suporte e recursos > Recursos e ferramentas > Clientes e chaves de API.
  3. Encontre o cliente de API que você criou e verifique se o escopo Fluxos de eventos mostra acesso de Leitura.

Testar o acesso à API

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

    CLIENT_ID="<your-client-id>"
CLIENT_SECRET="<your-client-secret>"
BASE_URL="https://api.crowdstrike.com"

# Get OAuth2 token
TOKEN=$(curl -s -X POST "${BASE_URL}/oauth2/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

# Test Event Streams API access (discover available data feeds)
curl -s -H "Authorization: Bearer ${TOKEN}" \
  "${BASE_URL}/sensors/entities/datafeed/v2?appId=secops-test&format=flatjson"

Uma resposta bem-sucedida retorna um objeto JSON que contém resources com os campos dataFeedURL e sessionToken para cada partição de stream disponível.

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 cs-stream-collector-sa
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect CrowdStrike Event Streams 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, cs-stream-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, cs-stream-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 cs-stream-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 se conectar à API Event Streams do CrowdStrike, consumir eventos 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 cs-stream-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 cs-stream-trigger.
    4. Clique em Salvar.

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

    1. Selecione Exigir autenticação.
    2. Confira o Identity and Access Management (IAM).

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

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione cs-stream-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 cs-stream-logs Nome do bucket do GCS
    GCS_PREFIX crowdstrike/stream Prefixo para arquivos de registro
    STATE_KEY crowdstrike/stream/state.json Caminho do arquivo de estado
    CS_BASE_URL https://api.crowdstrike.com URL de base da API CrowdStrike
    CS_CLIENT_ID your-client-id ID do cliente da API
    CS_CLIENT_SECRET your-client-secret Chave secreta do cliente da API
    CS_APP_ID secops-stream-collector ID exclusivo do app para fluxos de eventos (máximo de 32 caracteres alfanuméricos)
    STREAM_TIMEOUT 300 Tempo limite de leitura de stream em segundos
    MAX_RECORDS 10000 Máximo de registros por execução

  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 1 (apenas um consumidor de stream deve ser executado por vez).

  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 no campo Ponto de entrada.

  2. No editor de código inline, crie dois arquivos:

    • main.py:

      import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone

# 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', 'crowdstrike/stream')
STATE_KEY = os.environ.get('STATE_KEY', 'crowdstrike/stream/state.json')
CS_BASE_URL = os.environ.get('CS_BASE_URL', 'https://api.crowdstrike.com')
CS_CLIENT_ID = os.environ.get('CS_CLIENT_ID')
CS_CLIENT_SECRET = os.environ.get('CS_CLIENT_SECRET')
CS_APP_ID = os.environ.get('CS_APP_ID', 'secops-stream-collector')
STREAM_TIMEOUT = int(os.environ.get('STREAM_TIMEOUT', '300'))
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))

def get_oauth_token(base_url: str, client_id: str, client_secret: str) -> str:
    """Get CrowdStrike OAuth2 access token."""
    token_url = f"{base_url}/oauth2/token"
    body = f"client_id={client_id}&client_secret={client_secret}"
    response = http.request(
        'POST', token_url,
        body=body.encode('utf-8'),
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )
    if response.status != 200:
        raise Exception(f"OAuth token request failed: {response.status}")
    data = json.loads(response.data.decode('utf-8'))
    return data['access_token']

def refresh_stream_session(base_url: str, token: str, app_id: str, partition: int = 0):
    """Refresh an active stream session to keep it alive."""
    refresh_url = (
        f"{base_url}/sensors/entities/datafeed-actions/v1/{partition}"
        f"?appId={app_id}&action_name=refresh_active_stream_session"
    )
    response = http.request(
        'POST', refresh_url,
        headers={
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json',
        }
    )
    if response.status == 200:
        print("Stream session refreshed successfully")
    else:
        print(f"Warning: Stream session refresh returned {response.status}")

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch CrowdStrike
    Event Streams data and write to GCS.
    """

    if not all([GCS_BUCKET, CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state
        state = load_state(bucket, STATE_KEY)

        # Get OAuth token
        token = get_oauth_token(CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET)
        headers = {
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
        }

        # Discover available data feeds
        discover_url = (
            f"{CS_BASE_URL}/sensors/entities/datafeed/v2"
            f"?appId={CS_APP_ID}&format=flatjson"
        )
        response = http.request('GET', discover_url, headers=headers)

        if response.status != 200:
            print(f"Failed to discover data feed: {response.status}")
            print(f"Response: {response.data.decode('utf-8')}")
            return

        feed_data = json.loads(response.data.decode('utf-8'))
        resources = feed_data.get('resources', [])

        if not resources:
            print("No data feed resources available")
            return

        feed_url = resources[0].get('dataFeedURL', '')
        session_token = resources[0].get(
            'sessionToken', {}
        ).get('token', '')

        if not feed_url or not session_token:
            print("Missing feed URL or session token")
            return

        # Build stream URL with offset if available
        stream_url = feed_url
        offset = state.get('offset')
        if offset:
            stream_url = f"{feed_url}&offset={offset}"

        # Connect to Event Stream
        stream_headers = {
            'Authorization': f'Token {session_token}',
            'Accept': 'application/json',
            'Connection': 'keep-alive',
        }

        stream_http = urllib3.PoolManager(
            timeout=urllib3.Timeout(
                connect=10.0, read=float(STREAM_TIMEOUT)
            ),
            retries=False,
        )

        print(f"Connecting to Event Stream...")
        resp = stream_http.request(
            'GET', stream_url,
            headers=stream_headers,
            preload_content=False
        )

        if resp.status != 200:
            print(f"Stream connection failed: {resp.status}")
            return

        records = []
        latest_offset = offset
        now = datetime.now(timezone.utc)

        for line in resp.stream(4096):
            decoded = line.decode('utf-8').strip()
            if not decoded:
                continue

            try:
                event = json.loads(decoded)
                records.append(event)

                event_offset = event.get(
                    'metadata', {}
                ).get('offset')
                if event_offset is not None:
                    latest_offset = event_offset

                if len(records) >= MAX_RECORDS:
                    print(
                        f"Reached max records limit ({MAX_RECORDS})"
                    )
                    break
            except json.JSONDecodeError:
                continue

        resp.release_conn()

        if not records:
            print("No new events found in stream.")
            # Refresh stream session to keep it alive
            refresh_stream_session(
                CS_BASE_URL, token, CS_APP_ID
            )
            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(r, ensure_ascii=False) for r in records]
        ) + '\n'
        blob.upload_from_string(
            ndjson, content_type='application/x-ndjson'
        )

        print(
            f"Wrote {len(records)} records to "
            f"gs://{GCS_BUCKET}/{object_key}"
        )

        # Save state with latest offset
        new_state = {
            'offset': latest_offset,
            'last_run': now.isoformat()
        }
        save_state(bucket, STATE_KEY, new_state)

        # Refresh stream session to keep it alive
        refresh_stream_session(
            CS_BASE_URL, token, CS_APP_ID
        )

        print(f"Successfully processed {len(records)} events")

    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, state: dict):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: {state}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

    requirements.txt:

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

  4. Clique em Criar.

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job.
  2. Clique em Forçar execução para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique em cs-stream-collector.
  6. Clique na guia Registros.

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

    Connecting to Event Stream...
Wrote X records to gs://cs-stream-logs/crowdstrike/stream/logs_YYYYMMDD_HHMMSS.ndjson
Stream session refreshed successfully
Successfully processed X events

  8. Acesse Cloud Storage > Buckets.

  9. Clique em cs-stream-logs.

  10. Navegue até a pasta crowdstrike/stream/.

  11. Verifique se um novo arquivo .ndjson foi 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
  • HTTP 403: verifique se o cliente de API tem o escopo Fluxos de eventos: leitura
  • HTTP 429: limitação de taxa. Ajuste a frequência do programador ou aumente STREAM_TIMEOUT.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.
  • Nenhum recurso de feed de dados disponível: verifique se CS_APP_ID é exclusivo e não excede 32 caracteres alfanuméricos.

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, CrowdStrike Event Streams logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione Streaming do CrowdStrike Falcon como o Tipo de registro.
  7. Clique em Receber conta de serviço.

  8. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  9. Copie esse endereço de e-mail para usar na próxima etapa.

  10. Clique em Próxima.

  11. 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://cs-stream-logs/crowdstrike/stream/

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

    • 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

  12. Clique em Próxima.

  13. 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 em cs-stream-logs.
  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.

Precisa de mais ajuda?

Para mais informações sobre os feeds do Google Security Operations, consulte a documentação dos feeds do Google Security Operations. Para informações sobre os requisitos de cada tipo de feed, consulte Configuração de feed por tipo.

Para referência, consulte a documentação do analisador de fluxos de eventos do CrowdStrike Falcon.

Referência de delta de mapeamento de UDM

Em 15 de setembro de 2026, o Google SecOps lançou uma nova versão do analisador de fluxos de eventos do CrowdStrike Falcon, que inclui mudanças significativas no mapeamento dos campos de registro de fluxos de eventos do CrowdStrike Falcon para campos do UDM.

Delta de mapeamento de campo de registro

A tabela a seguir lista o delta de mapeamento para campos de registro de fluxos de eventos do CrowdStrike Falcon para UDM expostos antes de 15 de setembro de 2026 e posteriormente (listados nas colunas Mapeamento antigo e Novo mapeamento, respectivamente):

Campo de registro Mapeamento antigo Novo mapeamento Motivo para a alteração
deviceId additional.fields principal.asset.asset_id Mapeado para um campo do UDM mais específico em vez de rótulos.
event.AuditKeyValues (report_name) security_result.detection_fields target.file.names Mapeado para um campo do UDM mais específico em vez de rótulos.
event.IdpPolicyRuleName security_result.detection_fields security_result.rule_name Mapeado para um campo do UDM mais específico em vez de rótulos.
event.MitreAttack.TacticID security_result.detection_fields security_result.tactic.id Mapeado para um campo do UDM mais específico em vez de rótulos.
event.MitreAttack.TechniqueID security_result.detection_fields security_result.technique.id Mapeado para um campo do UDM mais específico em vez de rótulos.
event.NetworkAccesse.RemoteAddress principal.ip target.ip Atualizado porque event.NetworkAccesse.RemoteAddress representa o endereço do dispositivo de destino.
event.TacticId security_result.detection_fields security_result.tactic.id Mapeado para um campo do UDM mais específico em vez de rótulos.
event.TechniqueId security_result.detection_fields security_result.technique.id Mapeado para um campo do UDM mais específico em vez de rótulos.
event.UserId principal.user.userid principal.user.email_addresses,
principal.user.userid		 Se event.UserId for um endereço de e-mail, ele será mapeado para principal.user.email_addresses. Caso contrário, será mapeado para principal.user.userid.
remoteAddress principal.ip target.ip Atualizado porque remoteAddress representa o endereço do dispositivo de destino.
ruleName security_result.detection_fields security_result.rule_name Mapeado para um campo do UDM mais específico em vez de rótulos.
serviceName metadata.description,
target.application		 target.application O mapeamento redundante foi removido.

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.