Mengumpulkan log Layanan Perlindungan Identitas (IDP) CrowdStrike

Parser Version: 1.0

Didukung di:

Dokumen ini menjelaskan cara menyerap log Layanan Perlindungan Identitas (IDP) CrowdStrike ke Google Security Operations menggunakan Google Cloud Storage. Integrasi ini menggunakan CrowdStrike Unified Alerts API untuk mengumpulkan peristiwa Identity Protection dan menyimpannya dalam format NDJSON untuk diproses oleh parser CS_IDP bawaan.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

  • Instance Google SecOps
  • Project GCP dengan Cloud Storage API diaktifkan
  • Izin untuk membuat dan mengelola bucket GCS
  • Izin untuk mengelola kebijakan IAM di bucket GCS
  • Izin untuk membuat layanan Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
  • Akses istimewa ke Konsol CrowdStrike Falcon dan pengelolaan kunci API

Buat bucket Google Cloud Storage

  1. Buka Konsol Google Cloud.
  2. Pilih project Anda atau buat project baru.
  3. Di menu navigasi, buka Cloud Storage > Buckets.
  4. Klik Create bucket.

  5. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Beri nama bucket Anda Masukkan nama yang unik secara global (misalnya, crowdstrike-idp-logs-bucket)
    Location type Pilih berdasarkan kebutuhan Anda (Region, Dual-region, Multi-region)
    Location Pilih lokasi (misalnya, us-central1)
    Kelas penyimpanan Standar (direkomendasikan untuk log yang sering diakses)
    Kontrol akses Seragam (direkomendasikan)
    Alat perlindungan Opsional: Mengaktifkan pembuatan versi objek atau kebijakan retensi

  6. Klik Create.

Mendapatkan prasyarat CrowdStrike Identity Protection

  1. Login ke Konsol CrowdStrike Falcon.
  2. Buka Dukungan dan Sumber Daya > Klien dan kunci API.
  3. Klik Add new API Client.
  4. Berikan detail konfigurasi berikut:
    • Nama Klien: Masukkan Google SecOps IDP Integration.
    • Deskripsi: Masukkan API client for Google SecOps integration.
    • Cakupan: Pilih cakupan Alerts: READ (alerts:read) (ini mencakup pemberitahuan Identity Protection).
  5. Klik Tambahkan.
  6. Salin dan simpan detail berikut di lokasi yang aman:
    • Client ID
    • Rahasia Klien (hanya ditampilkan sekali)
    • URL dasar (contoh: api.crowdstrike.com untuk US-1, api.us-2.crowdstrike.com untuk US-2, api.eu-1.crowdstrike.com untuk EU-1)

Membuat akun layanan untuk fungsi Cloud Run

Cloud Run Function memerlukan akun layanan dengan izin untuk menulis ke bucket GCS.

Membuat akun layanan

  1. Di Konsol GCP, buka IAM & Admin > Service Accounts.
  2. Klik Create Service Account.
  3. Berikan detail konfigurasi berikut:
    • Nama akun layanan: Masukkan crowdstrike-idp-collector-sa.
    • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect CrowdStrike IDP logs.
  4. Klik Create and Continue.
  5. Di bagian Grant this service account access to project:
    1. Klik Pilih peran.
    2. Telusuri dan pilih Storage Object Admin.
    3. Klik + Add another role.
    4. Telusuri dan pilih Cloud Run Invoker.
    5. Klik + Add another role.
    6. Telusuri dan pilih Cloud Functions Invoker.
  6. Klik Lanjutkan.
  7. Klik Done.

Peran ini diperlukan untuk:

  • Storage Object Admin: Menulis log ke bucket GCS dan mengelola file status
  • Cloud Run Invoker: Izinkan Pub/Sub memanggil fungsi
  • Cloud Functions Invoker: Mengizinkan pemanggilan fungsi

Memberikan izin IAM di bucket GCS

Beri akun layanan izin tulis di bucket GCS:

  1. Buka Cloud Storage > Buckets.
  2. Klik nama bucket Anda.
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Tambahkan prinsipal: Masukkan email akun layanan (misalnya, crowdstrike-idp-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Tetapkan peran: Pilih Storage Object Admin.
  6. Klik Simpan.

Membuat topik Pub/Sub

Buat topik Pub/Sub yang akan dipublikasikan oleh Cloud Scheduler dan akan dilanggan oleh fungsi Cloud Run.

  1. Di Konsol GCP, buka Pub/Sub > Topics.
  2. Klik Create topic.
  3. Berikan detail konfigurasi berikut:
    • ID Topik: Masukkan crowdstrike-idp-trigger.
    • Biarkan setelan lainnya sesuai default.
  4. Klik Create.

Membuat fungsi Cloud Run untuk mengumpulkan log

Fungsi Cloud Run dipicu oleh pesan Pub/Sub dari Cloud Scheduler untuk mengambil log dari CrowdStrike Identity Protection API dan menuliskannya ke GCS.

  1. Di Konsol GCP, buka Cloud Run.
  2. Klik Create service.
  3. Pilih Function (gunakan editor edit langsung untuk membuat fungsi).

  4. Di bagian Konfigurasi, berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama layanan crowdstrike-idp-collector
    Region Pilih region yang cocok dengan bucket GCS Anda (misalnya, us-central1)
    Runtime Pilih Python 3.12 atau yang lebih baru

  5. Di bagian Trigger (optional):

    1. Klik + Tambahkan pemicu.
    2. Pilih Cloud Pub/Sub.
    3. Di Select a Cloud Pub/Sub topic, pilih topik crowdstrike-idp-trigger.
    4. Klik Simpan.

  6. Di bagian Authentication:

    1. Pilih Wajibkan autentikasi.
    2. Periksa Identity and Access Management (IAM).

  7. Scroll ke bawah dan luaskan Containers, Networking, Security.

  8. Buka tab Security:

    • Akun layanan: Pilih akun layanan crowdstrike-idp-collector-sa.

  9. Buka tab Containers:

    1. Klik Variables & Secrets.
    2. Klik + Add variable untuk setiap variabel lingkungan:
    Nama Variabel Nilai Contoh
    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 (opsional, maks. 10.000 per halaman)

  10. Scroll ke bawah di tab Variables & Secrets ke Requests:

    • Waktu tunggu permintaan: Masukkan 600 detik (10 menit).

  11. Buka tab Setelan di Penampung:

    • Di bagian Resources:
      • Memory: Pilih 512 MiB atau lebih tinggi.
      • CPU: Pilih 1.
    • Klik Done.

  12. Scroll ke Lingkungan eksekusi:

    • Pilih Default (direkomendasikan).

  13. Di bagian Revision scaling:

    • Minimum number of instances: Masukkan 0.
    • Maximum number of instances: Masukkan 100 (atau sesuaikan berdasarkan beban yang diharapkan).

  14. Klik Create.

  15. Tunggu hingga layanan dibuat (1-2 menit).

  16. Setelah layanan dibuat, editor kode inline akan terbuka secara otomatis.

Menambahkan kode fungsi

  1. Masukkan main di Function entry point

  2. Di editor kode inline, buat dua file:

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

  3. Klik Deploy untuk menyimpan dan men-deploy fungsi.

  4. Tunggu hingga deployment selesai (2-3 menit).

Buat tugas Cloud Scheduler

Cloud Scheduler memublikasikan pesan ke topik Pub/Sub secara berkala, sehingga memicu fungsi Cloud Run.

  1. Di Konsol GCP, buka Cloud Scheduler.
  2. Klik Create Job.

  3. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama crowdstrike-idp-collector-15m
    Region Pilih region yang sama dengan fungsi Cloud Run
    Frekuensi */15 * * * * (setiap 15 menit)
    Zona Waktu Pilih zona waktu (UTC direkomendasikan)
    Jenis target Pub/Sub
    Topik Pilih topik crowdstrike-idp-trigger
    Isi pesan {} (objek JSON kosong)

  4. Klik Create.

Menguji tugas penjadwal

  1. Di konsol Cloud Scheduler, temukan tugas Anda.
  2. Klik Jalankan paksa untuk memicu secara manual.
  3. Tunggu beberapa detik, lalu buka Cloud Run > Services > crowdstrike-idp-collector > Logs.
  4. Pastikan fungsi berhasil dieksekusi.
  5. Periksa bucket GCS untuk mengonfirmasi bahwa log telah ditulis.

Mengambil akun layanan Google SecOps

Google SecOps menggunakan akun layanan unik untuk membaca data dari bucket GCS Anda. Anda harus memberi akun layanan ini akses ke bucket Anda.

Dapatkan email akun layanan

  1. Buka SIEM Settings > Feeds.
  2. Klik Tambahkan Feed Baru.
  3. Klik Konfigurasi satu feed.
  4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, CrowdStrike Identity Protection Services logs).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih Crowdstrike Identity Protection Services sebagai Log type.

  7. Klik Get Service Account. Email akun layanan yang unik akan ditampilkan, misalnya:

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

  8. Salin alamat email ini untuk digunakan di langkah berikutnya.

Memberikan izin IAM ke akun layanan Google SecOps

Akun layanan Google SecOps memerlukan peran Storage Object Viewer di bucket GCS Anda.

  1. Buka Cloud Storage > Buckets.
  2. Klik nama bucket Anda.
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Add principals: Tempel email akun layanan Google SecOps.
    • Tetapkan peran: Pilih Storage Object Viewer.

  6. Klik Simpan.

Mengonfigurasi feed di Google SecOps untuk menyerap log layanan CrowdStrike Identity Protection Services

  1. Buka Setelan SIEM > Feed.
  2. Klik Tambahkan Feed Baru.
  3. Klik Konfigurasi satu feed.
  4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, CrowdStrike Identity Protection Services logs).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih Crowdstrike Identity Protection Services sebagai Log type.
  7. Klik Berikutnya.

  8. Tentukan nilai untuk parameter input berikut:

    • Storage bucket URL: Masukkan URI bucket GCS dengan jalur awalan:

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

      • Ganti:

        • crowdstrike-idp-logs-bucket: Nama bucket GCS Anda.
        • crowdstrike-idp/: Jalur folder/awalan tempat log disimpan.

    • Opsi penghapusan sumber: Pilih opsi penghapusan sesuai preferensi Anda:

      • Jangan pernah: Tidak pernah menghapus file apa pun setelah transfer (direkomendasikan untuk pengujian).
      • Hapus file yang ditransfer: Menghapus file setelah transfer berhasil.

      • Delete transferred files and empty directories: Menghapus file dan direktori kosong setelah transfer berhasil.

    • Usia File Maksimum: Menyertakan file yang diubah dalam beberapa hari terakhir. Defaultnya adalah 180 hari.

    • Namespace aset: Namespace aset.

    • Label penyerapan: Label yang akan diterapkan ke peristiwa dari feed ini.

  9. Klik Berikutnya.

  10. Tinjau konfigurasi feed baru Anda di layar Selesaikan, lalu klik Kirim.

Log Perubahan

Melihat Log Perubahan untuk parser ini

Perlu bantuan lain? Dapatkan jawaban dari anggota Komunitas dan profesional Google SecOps.