Raccogliere i log dei servizi CrowdStrike Identity Protection (IDP)

Parser Version: 1.0

Supportato in:

Questo documento spiega come importare i log dei servizi CrowdStrike Identity Protection (IDP) in Google Security Operations utilizzando Google Cloud Storage. L'integrazione utilizza l'API CrowdStrike Unified Alerts per raccogliere gli eventi di Identity Protection e li archivia in formato NDJSON per l'elaborazione da parte del parser CS_IDP integrato.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto GCP con l'API Cloud Storage abilitata
  • Autorizzazioni per creare e gestire bucket GCS
  • Autorizzazioni per gestire le policy IAM nei bucket GCS
  • Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso con privilegi alla console CrowdStrike Falcon e gestione delle chiavi API

Crea un bucket Google Cloud Storage

  1. Vai alla console Google Cloud.
  2. Seleziona il tuo progetto o creane uno nuovo.
  3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
  4. Fai clic su Crea bucket.

  5. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio crowdstrike-idp-logs-bucket).
    Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
    Località Seleziona la posizione (ad esempio, us-central1).
    Classe di archiviazione Standard (consigliato per i log a cui si accede di frequente)
    Controllo dell'accesso Uniforme (consigliato)
    Strumenti di protezione (Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Prerequisiti per CrowdStrike Identity Protection

  1. Accedi alla console CrowdStrike Falcon.
  2. Vai a Assistenza e risorse > Client e chiavi API.
  3. Fai clic su Aggiungi nuovo client API.
  4. Fornisci i seguenti dettagli di configurazione:
    • Nome client: inserisci Google SecOps IDP Integration.
    • Descrizione: inserisci API client for Google SecOps integration.
    • Ambiti: seleziona l'ambito Avvisi: LETTURA (alerts:read), che include gli avvisi di Identity Protection.
  5. Fai clic su Aggiungi.
  6. Copia e salva in una posizione sicura i seguenti dettagli:
    • ID client
    • Client secret (viene mostrato solo una volta)
    • URL di base (esempi: api.crowdstrike.com per US-1, api.us-2.crowdstrike.com per US-2, api.eu-1.crowdstrike.com per EU-1)

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni di scrittura nel bucket GCS.

Crea service account

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service account.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome service account: inserisci crowdstrike-idp-collector-sa.
    • Descrizione del service account: inserisci Service account for Cloud Run function to collect CrowdStrike IDP logs.
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo account di servizio l'accesso al progetto:
    1. Fai clic su Seleziona un ruolo.
    2. Cerca e seleziona Storage Object Admin.
    3. Fai clic su + Aggiungi un altro ruolo.
    4. Cerca e seleziona Cloud Run Invoker.
    5. Fai clic su + Aggiungi un altro ruolo.
    6. Cerca e seleziona Cloud Functions Invoker.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono obbligatori per:

  • Storage Object Admin: scrive i log nel bucket GCS e gestisce i file di stato
  • Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
  • Cloud Functions Invoker: consente la chiamata di funzioni

Concedi autorizzazioni IAM sul bucket GCS

Concedi al account di servizio le autorizzazioni di scrittura sul bucket GCS:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'email del account di servizio (ad esempio, crowdstrike-idp-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea argomento Pub/Sub

Crea un argomento Pub/Sub in cui Cloud Scheduler pubblicherà i messaggi e a cui la funzione Cloud Run si iscriverà.

  1. Nella console Google Cloud, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci crowdstrike-idp-trigger.
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API CrowdStrike Identity Protection e li scrive in GCS.

  1. Nella console Google Cloud, vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione (utilizza un editor in linea per creare una funzione).

  4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio crowdstrike-idp-collector
    Regione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive.

  5. Nella sezione Trigger (facoltativo):

    1. Fai clic su + Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento crowdstrike-idp-trigger.
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Controlla Identity and Access Management (IAM).

  7. Scorri verso il basso ed espandi Container, networking, sicurezza.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio crowdstrike-idp-collector-sa.

  9. Vai alla scheda Container:

    1. Fai clic su Variabili e secret.
    2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
    Nome variabile Valore di esempio
    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 (facoltativo, massimo 10.000 per pagina)

  10. Scorri verso il basso nella scheda Variabili e secret fino a Richieste:

    • Timeout della richiesta: inserisci 600 secondi (10 minuti).

  11. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
    • Fai clic su Fine.

  12. Scorri fino a Ambiente di esecuzione:

    • Seleziona Predefinita (opzione consigliata).

  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0.
    • Numero massimo di istanze: inserisci 100 (o modifica in base al carico previsto).

  14. Fai clic su Crea.

  15. Attendi la creazione del servizio (1-2 minuti).

  16. Dopo aver creato il servizio, si apre automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

  1. Inserisci main in Entry point della funzione

  2. Nell'editor di codice incorporato, crea due file:

    • Primo file: 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')
    • Secondo file: requirements.txt::
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

  4. Attendi il completamento del deployment (2-3 minuti).

Crea job Cloud Scheduler

Cloud Scheduler pubblica messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

  1. Nella console Google Cloud, vai a Cloud Scheduler.
  2. Fai clic su Crea job.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome crowdstrike-idp-collector-15m
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza */15 * * * * (ogni 15 minuti)
    Fuso orario Seleziona il fuso orario (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento crowdstrike-idp-trigger
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Testa il job di pianificazione

  1. Nella console Cloud Scheduler, trova il tuo job.
  2. Fai clic su Forza esecuzione per attivarlo manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > crowdstrike-idp-collector > Log.
  4. Verifica che la funzione sia stata eseguita correttamente.
  5. Controlla il bucket GCS per verificare che i log siano stati scritti.

Recuperare il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal tuo bucket GCS. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'email del account di servizio

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, CrowdStrike Identity Protection Services logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Crowdstrike Identity Protection Services come Tipo di log.

  7. Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email del account di servizio univoco, ad esempio:

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

  8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel tuo bucket GCS.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'email del account di servizio Google SecOps.
    • Assegna ruoli: seleziona Visualizzatore oggetti Storage.

  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i log del servizio di CrowdStrike Identity Protection Services

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed di notizie.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, CrowdStrike Identity Protection Services logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Crowdstrike Identity Protection Services come Tipo di log.
  7. Fai clic su Avanti.

  8. Specifica i valori per i seguenti parametri di input:

    • URL del bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:

      gs://crowdstrike-idp-logs-bucket/crowdstrike-idp/

      • Sostituisci:

        • crowdstrike-idp-logs-bucket: il nome del tuo bucket GCS.
        • crowdstrike-idp/: il percorso del prefisso/della cartella in cui sono archiviati i log.

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
      • Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.

      • Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.

    • Età massima file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.

  9. Fai clic su Avanti.

  10. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Log delle modifiche

Visualizza il log delle modifiche per questo parser

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.