收集 CrowdStrike FileVantage 日志
解析器版本：1.0
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Google Cloud Storage 将 CrowdStrike FileVantage 日志注入到 Google Security Operations。CrowdStrike FileVantage 是一种文件完整性监控解决方案，可跟踪整个环境中关键文件和目录的变化。
准备工作
确保您满足以下前提条件：
- Google SecOps 实例
- 启用了 Cloud Storage API 的 GCP 项目
- 创建和管理 GCS 存储桶的权限
- 管理 GCS 存储桶的 IAM 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
对 CrowdStrike Falcon 控制台的特权访问权限
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建一个新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储桶。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 crowdstrike-filevantage-logs）位置类型 根据您的需求进行选择（区域级、双区域级、多区域级） 位置 选择位置（例如 us-central1）存储类别 标准（建议用于频繁访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
收集 CrowdStrike FileVantage API 凭据
- 登录 CrowdStrike Falcon 控制台。
- 依次前往支持和资源 > API 客户端和密钥。
- 点击添加新的 API 客户端。
- 提供以下配置详细信息：
- 客户端名称：输入一个描述性名称（例如
Google SecOps FileVantage Integration）。 - 说明：输入集成用途的简要说明。
- API 范围：选择 Falcon FileVantage:read。
- 客户端名称：输入一个描述性名称（例如
- 点击 Add 完成此过程。
复制以下详细信息并将其保存在安全的位置：
- 客户端 ID (Client ID)
- 客户端密钥 (Client Secret)
- 基础网址（决定您的云区域）
验证权限
如需验证账号是否具有所需权限，请执行以下操作：
- 登录 CrowdStrike Falcon 控制台。
- 依次前往支持和资源 > API 客户端和密钥。
- 如果您可以查看 API 客户端和密钥页面并创建新的 API 客户端，则表示您拥有所需的权限。
- 如果您无法访问此页面，请与 CrowdStrike 管理员联系，让其授予您 Falcon 管理员角色。
测试 API 访问权限
在继续进行集成之前，请测试您的凭据：
# Replace with your actual credentials FALCON_CLIENT_ID="your-client-id" FALCON_CLIENT_SECRET="your-client-secret" FALCON_BASE_URL="https://api.crowdstrike.com" # Get OAuth token TOKEN=$(curl -s -X POST "${FALCON_BASE_URL}/oauth2/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "client_id=${FALCON_CLIENT_ID}&client_secret=${FALCON_CLIENT_SECRET}&grant_type=client_credentials" \ | grep -o '"access_token":"[^"]*' | cut -d'"' -f4) # Test FileVantage API access curl -v -H "Authorization: Bearer ${TOKEN}" \ "${FALCON_BASE_URL}/filevantage/queries/changes/v3?limit=1"
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有写入 GCS 存储桶和被 Pub/Sub 调用的权限。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
crowdstrike-filevantage-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect CrowdStrike FileVantage logs。
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
以下操作需要这些角色：
- Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储桶的 IAM 权限
向服务账号授予对 GCS 存储桶的写入权限：
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称（例如
crowdstrike-filevantage-logs）。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
crowdstrike-filevantage-sa@PROJECT_ID.iam.gserviceaccount.com）。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
crowdstrike-filevantage-trigger。 - 将其他设置保留为默认值。
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发，用于从 CrowdStrike FileVantage API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 crowdstrike-filevantage-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择 Pub/Sub 主题 (
crowdstrike-filevantage-trigger)。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号 (
crowdstrike-filevantage-sa)。
- 服务账号：选择服务账号 (
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量：
变量名称 示例值 GCS_BUCKETcrowdstrike-filevantage-logsGCS_PREFIXfilevantage/STATE_KEYfilevantage/state.jsonFALCON_CLIENT_IDyour-client-idFALCON_CLIENT_SECRETyour-client-secretFALCON_BASE_URLhttps://api.crowdstrike.com（美国 1）/https://api.us-2.crowdstrike.com（美国 2）/https://api.eu-1.crowdstrike.com（欧盟 1）在变量和密钥部分中，向下滚动到请求：
- 请求超时：输入
600秒（10 分钟）。
- 请求超时：输入
前往设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值。
- CPU：选择 1。
- 在资源部分中：
在修订版本伸缩部分中：
- 实例数下限：输入
0。 - 实例数上限：输入
100（或根据预期负载进行调整）。
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点中输入 main
在内嵌代码编辑器中，创建两个文件：
- 第一个文件：main.py:：
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time # 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() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch CrowdStrike FileVantage logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'filevantage/') state_key = os.environ.get('STATE_KEY', 'filevantage/state.json') client_id = os.environ.get('FALCON_CLIENT_ID') client_secret = os.environ.get('FALCON_CLIENT_SECRET') base_url = os.environ.get('FALCON_BASE_URL') if not all([bucket_name, client_id, client_secret, base_url]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Get OAuth token token_url = f"{base_url}/oauth2/token" token_headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } token_data = f"client_id={client_id}&client_secret={client_secret}&grant_type=client_credentials" token_response = http.request( 'POST', token_url, body=token_data.encode('utf-8'), headers=token_headers ) if token_response.status != 200: print(f"Failed to get OAuth token: {token_response.status}") print(f"Response: {token_response.data.decode('utf-8')}") return token_data_json = json.loads(token_response.data.decode('utf-8')) access_token = token_data_json['access_token'] # Get last checkpoint last_timestamp = get_last_checkpoint(bucket, state_key) # Fetch file changes using v3 endpoint (high volume query) changes_url = f"{base_url}/filevantage/queries/changes/v3" headers = { 'Authorization': f'Bearer {access_token}', 'Accept': 'application/json' } # Build query parameters params = [] params.append('limit=5000') params.append('sort=action_timestamp|asc') if last_timestamp: params.append(f"filter=action_timestamp:>'{last_timestamp}'") query_url = f"{changes_url}?{'&'.join(params)}" backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('GET', query_url, headers=headers) if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue if response.status != 200: print(f"Failed to query changes: {response.status}") print(f"Response: {response.data.decode('utf-8')}") return break else: print("Max retries exceeded") return response_data = json.loads(response.data.decode('utf-8')) change_ids = response_data.get('resources', []) if not change_ids: print("No new changes found") return # Get detailed change information using v2 endpoint details_url = f"{base_url}/filevantage/entities/changes/v2" batch_size = 500 all_changes = [] latest_timestamp = last_timestamp for i in range(0, len(change_ids), batch_size): batch_ids = change_ids[i:i + batch_size] # Build query string with multiple ids parameters ids_params = '&'.join([f'ids={id}' for id in batch_ids]) details_query_url = f"{details_url}?{ids_params}" backoff = 1.0 for attempt in range(max_retries): details_response = http.request('GET', details_query_url, headers=headers) if details_response.status == 429: retry_after = int(details_response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue if details_response.status == 200: details_data = json.loads(details_response.data.decode('utf-8')) changes = details_data.get('resources', []) all_changes.extend(changes) # Track latest timestamp for change in changes: change_time = change.get('action_timestamp') if change_time and (not latest_timestamp or change_time > latest_timestamp): latest_timestamp = change_time break else: print(f"Failed to get change details (batch {i//batch_size + 1}): {details_response.status}") break if all_changes: # Store logs in GCS timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') blob_name = f"{prefix}filevantage_changes_{timestamp}.json" blob = bucket.blob(blob_name) log_lines = '\n'.join(json.dumps(change) for change in all_changes) blob.upload_from_string(log_lines, content_type='application/json') # Update checkpoint save_checkpoint(bucket, state_key, latest_timestamp) print(f"Stored {len(all_changes)} changes in GCS: {blob_name}") except Exception as e: print(f"Error: {str(e)}") raise def get_last_checkpoint(bucket, key): """Get the 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') except Exception as e: print(f"Error reading checkpoint: {e}") return None def save_checkpoint(bucket, key, timestamp): """Save the last processed timestamp to GCS state file""" try: state = { 'last_timestamp': timestamp, 'updated_at': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f"Error saving checkpoint: {e}")- 第二个文件：requirements.txt:：
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署函数。
等待部署完成（2-3 分钟）。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP 控制台中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 crowdstrike-filevantage-hourly区域 选择与 Cloud Run 函数相同的区域 频率 */15 * * * *（每 15 分钟）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择 Pub/Sub 主题 ( crowdstrike-filevantage-trigger)消息正文 {}（空 JSON 对象）点击创建。
测试调度器作业
- 在 Cloud Scheduler 控制台中，找到您的作业 (
crowdstrike-filevantage-hourly)。 - 点击强制运行以手动触发。
- 等待几秒钟，然后依次前往 Cloud Run > 服务 > crowdstrike-filevantage-collector > 日志。
- 验证函数是否已成功执行。
- 检查 GCS 存储桶 (
crowdstrike-filevantage-logs) 以确认日志是否已写入。
检索 Google SecOps 服务账号
Google SecOps 使用唯一的服务账号从您的 GCS 存储桶中读取数据。您必须向此服务账号授予对您的存储桶的访问权限。
获取服务账号电子邮件地址
- 前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
CrowdStrike FileVantage logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 CrowdStrike Filevantage 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称（例如
crowdstrike-filevantage-logs）。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。
在 Google SecOps 中配置 Feed 以注入 CrowdStrike FileVantage 日志
- 前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
CrowdStrike FileVantage logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 CrowdStrike Filevantage 作为日志类型。
- 点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://crowdstrike-filevantage-logs/filevantage/将
crowdstrike-filevantage-logs：您的 GCS 存储桶名称。filevantage/：存储日志的前缀/文件夹路径。
来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：成功转移后删除文件。
删除已转移的文件和空目录：成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改过的文件。默认值为 180 天。
资源命名空间：资源命名空间。
注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。