收集 CrowdStrike Identity Protection (IDP) Services 日志

解析器版本： 1.0

支持：

Google SecOps SIEM

本文档介绍如何使用 Google Cloud Storage 将 CrowdStrike Identity Protection (IDP) Services 日志注入 Google Security Operations。该集成使用 CrowdStrike Unified Alerts API 收集 Identity Protection 事件，并以 NDJSON 格式存储这些事件，以便由内置的 CS_IDP 解析器进行处理。

准备工作

请确保满足以下前提条件：

• Google SecOps 实例
• 启用了 Cloud Storage API 的 GCP 项目
• 创建和管理 GCS 存储分区的权限
• 管理 GCS 存储分区的 IAM 政策的权限
• 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
• 对 CrowdStrike Falcon 控制台和 API 密钥管理的特权访问权限

创建 Google Cloud Storage 存储桶

1. 前往 Google Cloud 控制台。
2. 选择您的项目或创建一个新项目。
3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
4. 点击创建存储桶 。

5. 提供以下配置详细信息：

   设置	值
   指定存储桶的名称	输入一个全局唯一的名称（例如 crowdstrike-idp-logs-bucket）
   位置类型	根据您的需求进行选择（区域、双区域、多区域）
   位置	选择位置（例如 us-central1）
   存储类别	标准（建议用于经常访问的日志）
   访问权限控制	统一（推荐）
   保护工具	可选：启用对象版本控制或保留政策

6. 点击创建 。

获取 CrowdStrike Identity Protection 前提条件

1. 登录 CrowdStrike Falcon 控制台 。
2. 依次前往支持和资源 > API 客户端和密钥。
3. 点击 Add new API Client （添加新的 API 客户端）。
4. 提供以下配置详细信息：
   • Client Name（客户端名称）：输入 Google SecOps IDP Integration。
   • Description（说明）：输入 API client for Google SecOps integration。
   • Scopes（范围）：选择 Alerts: READ（提醒：读取）范围（包括 Identity Protection 提醒）。
5. 点击 Add （添加）。
6. 复制以下详细信息并将其保存在安全的位置：
   • Client ID （客户端 ID）
   • Client Secret （客户端密钥）（此密钥仅显示一次）
   • Base 网址 （基本网址）（示例：美国 1 区为 api.crowdstrike.com，美国 2 区为 api.us-2.crowdstrike.com，欧盟 1 区为 api.eu-1.crowdstrike.com）

为 Cloud Run 函数创建服务帐号

Cloud Run 函数需要一个具有写入 GCS 存储桶权限的服务帐号。

创建服务帐号

1. 在 GCP 控制台 中，依次前往 IAM 和管理 > 服务账号。
2. 点击创建服务账号 。
3. 提供以下配置详细信息：
   • Service account name（服务账号名称）：输入 crowdstrike-idp-collector-sa。
   • Service account description（服务账号说明）：输入 Service account for Cloud Run function to collect CrowdStrike IDP logs。
4. 点击创建并继续 。
5. 在向此服务帐号授予对项目的访问权限 部分：
   1. 点击选择角色 。
   2. 搜索并选择 Storage Object Admin （存储对象管理员）。
   3. 点击 + 添加其他角色。
   4. 搜索并选择 Cloud Run Invoker （Cloud Run 调用方）。
   5. 点击 + 添加其他角色。
   6. 搜索并选择 Cloud Functions Invoker （Cloud Functions 调用方）。
6. 点击继续 。
7. 点击完成 。

必须拥有以下角色，才能：

• Storage Object Admin（存储对象管理员）：将日志写入 GCS 存储桶并管理状态文件
• Cloud Run Invoker （Cloud Run 调用方）：允许 Pub/Sub 调用该函数
• Cloud Functions Invoker（Cloud Functions 调用方）：允许函数调用

授予对 GCS 存储桶的 IAM 权限

向服务帐号授予对 GCS 存储桶的写入权限：

1. 依次前往 Cloud Storage > 存储分区。
2. 点击您的存储桶名称。
3. 前往权限 标签页。
4. 点击授予访问权限 。
5. 提供以下配置详细信息：
   • 添加主账号：输入服务帐号电子邮件地址（例如 crowdstrike-idp-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。
   • 分配角色：选择 Storage Object Admin（存储对象管理员）。
6. 点击保存 。

创建 Pub/Sub 主题

创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。

1. 在 GCP 控制台 中，依次前往 Pub/Sub > 主题。
2. 点击创建主题 。
3. 提供以下配置详细信息：
   • 主题 ID：输入 crowdstrike-idp-trigger。
   • 将其他设置保留为默认设置。
4. 点击创建 。

创建 Cloud Run 函数以收集日志

Cloud Run 函数由 Cloud Scheduler 中的 Pub/Sub 消息触发，以从 CrowdStrike Identity Protection API 中提取日志并将其写入 GCS。

1. 在 GCP 控制台 中，前往 Cloud Run。
2. 点击创建服务 。
3. 选择函数 （使用内嵌编辑器创建函数）。

4. 在配置 部分中，提供以下配置详细信息：

   设置	值
   Service 名称	crowdstrike-idp-collector
   区域	选择与您的 GCS 存储桶匹配的区域（例如 us-central1）
   运行时	选择 Python 3.12 或更高版本

5. 在触发器（可选） 部分中：

   1. 点击 + 添加触发器。
   2. 选择 Cloud Pub/Sub 。
   3. 在选择 Cloud Pub/Sub 主题 中，选择主题 crowdstrike-idp-trigger。
   4. 点击保存 。

6. 在身份验证 部分中：

   1. 选择需要进行身份验证 。
   2. 勾选 Identity and Access Management (IAM) （身份和访问权限管理 (IAM)）。
   注意： Pub/Sub 在调用函数时会自动处理身份验证。

7. 向下滚动并展开容器、网络、安全性 。

8. 前往安全性 标签页：

   • 服务账号：选择服务账号 crowdstrike-idp-collector-sa。

9. 前往容器 标签页：

   1. 点击变量和密钥 。
   2. 为每个环境变量点击 + 添加变量 ：

   变量名称	示例值
   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（可选，每页最多 10000 个）

10. 在变量和密钥 标签页中向下滚动到 Requests（请求）：

    • Request timeout（请求超时）：输入 600 秒（10 分钟）。

11. 前往容器 中的设置 标签页：

    • 在资源 部分中：
      • 内存：选择 512 MiB 或更高。
      • CPU：选择 1。
    • 点击完成 。

12. 滚动到执行环境：

    • 选择默认 （推荐）。

13. 在修订版本伸缩 部分中：

    • 最小实例数：输入 0。
    • 实例数上限：输入 100（或根据预期负载进行调整）。

14. 点击创建 。

15. 等待服务创建完成（1-2 分钟）。

16. 服务创建完成后，系统会自动打开内嵌代码编辑器 。

添加函数代码

1. 在函数入口点 中输入 main

2. 在内嵌代码编辑器中，创建两个文件：

   • 第一个文件：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')
   

   • 第二个文件：requirements.txt:：

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. 点击部署 以保存和部署函数。

4. 等待部署完成（2-3 分钟）。

创建 Cloud Scheduler 作业

Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。

1. 在 GCP 控制台 中，前往 Cloud Scheduler。
2. 点击创建作业 。

3. 提供以下配置详细信息：

   设置	值
   名称	crowdstrike-idp-collector-15m
   区域	选择与 Cloud Run 函数相同的区域
   频率	*/15 * * * *（每 15 分钟一次）
   时区	选择时区（建议使用 UTC）
   定位类型	Pub/Sub
   主题	选择主题 crowdstrike-idp-trigger
   消息正文	{}（空 JSON 对象）

4. 点击创建 。

测试调度器作业

1. 在 Cloud Scheduler 控制台中，找到您的作业。
2. 点击强制运行 以手动触发。
3. 等待几秒钟，然后前往 Cloud Run > 服务 > crowdstrike-idp-collector > 日志。
4. 验证函数是否已成功执行。
5. 检查 GCS 存储桶以确认日志已写入。

检索 Google SecOps 服务帐号

Google SecOps 使用唯一的服务帐号从您的 GCS 存储桶中读取数据。您必须向此服务帐号授予对您的存储桶的访问权限。

获取服务帐号电子邮件地址

1. 依次前往 SIEM 设置 > Feed。
2. 点击 Add New Feed （添加新 Feed）。
3. 点击配置单个 Feed 。
4. 在 Feed 名称 字段中，输入 Feed 的名称（例如 CrowdStrike Identity Protection Services logs）。
5. 选择 Google Cloud Storage V2 作为来源类型 。
6. 选择 Crowdstrike Identity Protection Services 作为日志类型 。

7. 点击 Get Service Account （获取服务账号）。系统会显示一个唯一的服务帐号电子邮件地址，例如：

   [e-mail]
   

8. 复制此电子邮件地址，以便在下一步中使用。

   注意： 每个 Google SecOps 实例都有一个唯一的服务帐号。请勿使用其他文档或示例中的服务账号。

向 Google SecOps 服务帐号授予 IAM 权限

Google SecOps 服务帐号需要对您的 GCS 存储桶拥有 Storage Object Viewer （存储对象查看者）角色。

1. 依次前往 Cloud Storage > 存储分区。
2. 点击您的存储桶名称。
3. 前往权限 标签页。
4. 点击授予访问权限 。
5. 提供以下配置详细信息：
   • 添加主账号：粘贴 Google SecOps 服务帐号电子邮件地址。
   • 分配角色：选择 Storage Object Viewer（存储对象查看者）。

6. 点击保存 。

   注意： 如果您打算使用删除选项“删除已转移的文件”或“删除已转移的文件和空目录”，请授予 Storage Object Admin （存储对象管理员）角色，而不是 Storage Object Viewer。

在 Google SecOps 中配置 Feed 以注入 CrowdStrike Identity Protection Services 日志

1. 依次前往 SIEM 设置 > Feed。
2. 点击 Add New Feed （添加新 Feed）。
3. 点击配置单个 Feed 。
4. 在 Feed 名称 字段中，输入 Feed 的名称（例如 CrowdStrike Identity Protection Services logs）。
5. 选择 Google Cloud Storage V2 作为来源类型 。
6. 选择 Crowdstrike Identity Protection Services 作为日志类型 。
7. 点击下一步 。

8. 为以下输入参数指定值：

   • Storage bucket 网址（存储桶网址）：输入带有前缀路径的 GCS 存储桶 URI：

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

     • 将

       • crowdstrike-idp-logs-bucket 替换为您的 GCS 存储桶名称。
       • crowdstrike-idp/ 替换为存储日志的前缀/文件夹路径。
   注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

   • Source deletion option（来源删除选项）：根据您的偏好选择删除选项：

     • 永不：转移后永不删除任何文件（建议用于测试）。
     • 删除已转移的文件：成功转移后删除文件。

     • 删除已转移的文件和空目录：成功转移后删除文件和空目录。

       注意： 如果您选择删除选项，服务帐号必须拥有 Storage Object Admin （存储对象管理员）角色，而不是 Storage Object Viewer。请相应地更新 IAM 权限。

   • Maximum File Age（文件最长保留时间）：包含在过去指定天数内修改的文件。默认值为 180 天。

   • Asset namespace（资产命名空间）：资产命名空间。

   • 注入标签：要应用于此 Feed 中的事件的标签。

9. 点击下一步 。

10. 在最终确定 屏幕中查看新的 Feed 配置，然后点击提交 。

更新日志

查看此解析器的更新日志

需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。