收集 Cisco CloudLock CASB 日志
本文档介绍了如何使用 Google Cloud Storage 将 Cisco CloudLock CASB 日志提取到 Google Security Operations。解析器会从 JSON 日志中提取字段,然后将这些字段转换并映射到 Unified Data Model (UDM)。它负责处理日期解析、将特定字段转换为字符串、将字段映射到 UDM 实体(元数据、目标、安全结果、关于),并遍历匹配项以提取检测字段,最终将所有提取的数据合并到 @output 字段中。
Cisco CloudLock 是一款云原生 Cloud Access Security Broker (CASB),可提供对云应用的可见性和控制。它可帮助组织发现影子 IT、强制执行数据泄露防护政策、检测威胁,并确保 SaaS 应用符合相关法规。
准备工作
确保您满足以下前提条件:
- Google SecOps 实例
- 启用了 Cloud Storage API 的 GCP 项目
- 创建和管理 GCS 存储分区的权限
- 管理 GCS 存储分区的 IAM 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 对 Cisco CloudLock 管理控制台的特权访问权限
获取 Cisco CloudLock API 前提条件
如需开始使用,请与 Cloudlock 支持团队联系,以获取 Cloudlock API 网址。在 Cloudlock 应用中生成访问令牌,方法是选择“设置”页面中的“身份验证和 API”标签页,然后点击“生成”。
- 登录 Cisco CloudLock 管理控制台。
- 前往设置 > 身份验证和 API。
- 在 API 下,点击 Generate(生成)以创建访问令牌。
- 请复制以下详细信息并将其保存在安全的位置:
- API 访问令牌
- API 基本网址(由 Cisco CloudLock 支持团队提供,电子邮件地址为 [email protected])
创建 Google Cloud Storage 存储分区
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中,依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息:
设置 值 为存储分区命名 输入一个全局唯一的名称(例如 cisco-cloudlock-logs)位置类型 根据您的需求进行选择(区域级、双区域级、多区域级) 位置 选择相应位置(例如 us-central1)存储类别 标准(建议用于经常访问的日志) 访问权限控制 统一(推荐) 保护工具 可选:启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个有权写入 GCS 存储分区的服务账号。
创建服务账号
- 在 GCP 控制台中,依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息:
- 服务账号名称:输入
cloudlock-data-export-sa。 - 服务账号说明:输入
Service account for Cloud Run function to collect Cisco CloudLock logs。
- 服务账号名称:输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分:
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
授予对 GCS 存储分区的 IAM 权限
向服务账号授予对 GCS 存储分区的写入权限:
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:输入服务账号电子邮件地址(例如
cloudlock-data-export-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 分配角色:选择 Storage Object Admin。
- 添加主账号:输入服务账号电子邮件地址(例如
- 点击保存。
创建发布/订阅主题
创建一个 Pub/Sub 主题,供 Cloud Scheduler 发布消息,并供 Cloud Run 函数订阅。
- 在 GCP 控制台中,前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息:
- 主题 ID:输入
cloudlock-data-export-trigger。 - 将其他设置保留为默认值。
- 主题 ID:输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发,以从 Cisco CloudLock API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中,前往 Cloud Run。
- 点击创建服务。
- 选择函数(使用内嵌编辑器创建函数)。
在配置部分中,提供以下配置详细信息:
设置 值 Service 名称 cloudlock-data-export区域 选择与您的 GCS 存储分区匹配的区域(例如 us-central1)运行时 选择 Python 3.12 或更高版本 在触发器(可选)部分中:
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中,选择相应主题 (
cloudlock-data-export-trigger)。 - 点击保存。
在身份验证部分中:
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
滚动到容器、网络、安全性并展开该部分。
前往安全标签页:
- 服务账号:选择服务账号 (
cloudlock-data-export-sa)。
- 服务账号:选择服务账号 (
前往容器标签页:
- 点击变量和密钥。
为每个环境变量点击 + 添加变量:
变量名称 示例值 GCS_BUCKETcisco-cloudlock-logsGCS_PREFIXcloudlock/STATE_KEYcloudlock/state.jsonCLOUDLOCK_API_TOKENyour-api-tokenCLOUDLOCK_API_BASEhttps://api.cloudlock.com
在变量和密钥标签页中,向下滚动到请求:
- 请求超时:输入
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 http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from Cisco CloudLock API 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', 'cloudlock/') state_key = os.environ.get('STATE_KEY', 'cloudlock/state.json') api_token = os.environ.get('CLOUDLOCK_API_TOKEN') api_base = os.environ.get('CLOUDLOCK_API_BASE') if not all([bucket_name, api_token, api_base]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Load state (last processed offset for each endpoint) state = load_state(bucket, state_key) print(f'Processing logs with state: {state}') # Create Authorization header headers = { 'Authorization': f'Bearer {api_token}', 'Content-Type': 'application/json' } # Fetch incidents data (using offset-based pagination) incidents_offset = state.get('incidents_offset', 0) incidents, new_incidents_offset = fetch_cloudlock_incidents( http, api_base, headers, incidents_offset ) if incidents: upload_to_gcs_ndjson(bucket, prefix, 'incidents', incidents) print(f'Uploaded {len(incidents)} incidents to GCS') state['incidents_offset'] = new_incidents_offset # Fetch activities data (using time-based filtering with offset pagination) activities_last_time = state.get('activities_last_time') if not activities_last_time: activities_last_time = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat() activities_offset = state.get('activities_offset', 0) activities, new_activities_offset, newest_activity_time = fetch_cloudlock_activities( http, api_base, headers, activities_last_time, activities_offset ) if activities: upload_to_gcs_ndjson(bucket, prefix, 'activities', activities) print(f'Uploaded {len(activities)} activities to GCS') state['activities_offset'] = new_activities_offset if newest_activity_time: state['activities_last_time'] = newest_activity_time # Fetch entities data (using offset-based pagination) entities_offset = state.get('entities_offset', 0) entities, new_entities_offset = fetch_cloudlock_entities( http, api_base, headers, entities_offset ) if entities: upload_to_gcs_ndjson(bucket, prefix, 'entities', entities) print(f'Uploaded {len(entities)} entities to GCS') state['entities_offset'] = new_entities_offset # Update consolidated state state['updated_at'] = datetime.now(timezone.utc).isoformat() save_state(bucket, state_key, state) print('CloudLock data export completed successfully') except Exception as e: print(f'Error processing logs: {str(e)}') raise def make_api_request(http, url, headers, retries=3): """Make API request with exponential backoff retry logic.""" for attempt in range(retries): try: response = http.request('GET', url, headers=headers) if response.status == 200: return response elif response.status == 429: # Rate limit retry_after = int(response.headers.get('Retry-After', 60)) print(f'Rate limited, waiting {retry_after} seconds') time.sleep(retry_after) else: print(f'API request failed with status {response.status}: {response.data.decode("utf-8")}') except Exception as e: print(f'Request attempt {attempt + 1} failed: {str(e)}') if attempt < retries - 1: wait_time = 2 ** attempt time.sleep(wait_time) else: raise return None def fetch_cloudlock_incidents(http, api_base, headers, start_offset=0): """ Fetch incidents data from Cisco CloudLock API using offset-based pagination. Note: The CloudLock API does not support updated_after parameter. This function uses offset-based pagination. For production use, consider implementing time-based filtering using created_at or updated_at fields in the response data. """ url = f"{api_base}/api/v2/incidents" limit = 1000 offset = start_offset all_data = [] try: while True: # Build URL with parameters full_url = f"{url}?limit={limit}&offset={offset}" print(f"Fetching incidents with offset: {offset}") response = make_api_request(http, full_url, headers) if not response: break data = json.loads(response.data.decode('utf-8')) # CloudLock API returns items in 'items' array batch_data = data.get('items', []) if not batch_data: print("No more incidents to fetch") break all_data.extend(batch_data) # Check if we've reached the end total = data.get('total', 0) results = data.get('results', len(batch_data)) print(f"Fetched {results} incidents (total available: {total})") if results < limit or offset + results >= total: print("Reached end of incidents") break offset += limit print(f"Fetched {len(all_data)} total incidents") return all_data, offset except Exception as e: print(f"Error fetching incidents: {str(e)}") return [], start_offset def fetch_cloudlock_activities(http, api_base, headers, from_time, start_offset=0): """ Fetch activities data from Cisco CloudLock API using time-based filtering and offset pagination. """ url = f"{api_base}/api/v2/activities" limit = 1000 offset = start_offset all_data = [] newest_time = None try: while True: # Build URL with time filter and pagination full_url = f"{url}?limit={limit}&offset={offset}" print(f"Fetching activities with offset: {offset}") response = make_api_request(http, full_url, headers) if not response: break data = json.loads(response.data.decode('utf-8')) batch_data = data.get('items', []) if not batch_data: print("No more activities to fetch") break # Filter activities by time (client-side filtering since API may not support time parameters) filtered_batch = [] for item in batch_data: item_time = item.get('timestamp') or item.get('created_at') if item_time and item_time >= from_time: filtered_batch.append(item) if not newest_time or item_time > newest_time: newest_time = item_time all_data.extend(filtered_batch) results = data.get('results', len(batch_data)) total = data.get('total', 0) print(f"Fetched {results} activities, {len(filtered_batch)} after time filter (total available: {total})") if results < limit or offset + results >= total: print("Reached end of activities") break offset += limit print(f"Fetched {len(all_data)} total activities") return all_data, offset, newest_time except Exception as e: print(f"Error fetching activities: {str(e)}") return [], start_offset, None def fetch_cloudlock_entities(http, api_base, headers, start_offset=0): """ Fetch entities data from Cisco CloudLock API using offset-based pagination. Note: This endpoint requires the Entity Cache feature. If not enabled, use the incident entities endpoint as an alternative. """ url = f"{api_base}/api/v2/entities" limit = 1000 offset = start_offset all_data = [] try: while True: full_url = f"{url}?limit={limit}&offset={offset}" print(f"Fetching entities with offset: {offset}") response = make_api_request(http, full_url, headers) if not response: break data = json.loads(response.data.decode('utf-8')) batch_data = data.get('items', []) if not batch_data: print("No more entities to fetch") break all_data.extend(batch_data) results = data.get('results', len(batch_data)) total = data.get('total', 0) print(f"Fetched {results} entities (total available: {total})") if results < limit or offset + results >= total: print("Reached end of entities") break offset += limit print(f"Fetched {len(all_data)} total entities") return all_data, offset except Exception as e: print(f"Error fetching entities: {str(e)}") return [], start_offset def upload_to_gcs_ndjson(bucket, prefix, data_type, data): """Upload data to GCS bucket in NDJSON format (one JSON object per line).""" timestamp = datetime.now(timezone.utc).strftime('%Y/%m/%d/%H') filename = f"{prefix}{data_type}/{timestamp}/cloudlock_{data_type}_{int(datetime.now(timezone.utc).timestamp())}.jsonl" try: # Convert to NDJSON format ndjson_content = '\n'.join([json.dumps(item, separators=(',', ':')) for item in data]) blob = bucket.blob(filename) blob.upload_from_string( ndjson_content, content_type='application/x-ndjson' ) print(f"Successfully uploaded {filename} to GCS") except Exception as e: print(f"Error uploading to GCS: {str(e)}") raise def load_state(bucket, key): """Load state from GCS with separate tracking for each endpoint.""" 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: {str(e)}') print("No previous state found, starting fresh") return {} def save_state(bucket, key, state): """Save consolidated state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print("Updated state successfully") except Exception as e: print(f"Error updating state: {str(e)}") raise- 第二个文件:requirements.txt::
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
点击部署以保存并部署该函数。
等待部署完成(2-3 分钟)。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息,从而触发 Cloud Run 函数。
- 在 GCP Console 中,前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息:
设置 值 名称 cloudlock-data-export-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *(每小时一次,在整点时)时区 选择时区(建议选择世界协调时间 [UTC]) 目标类型 Pub/Sub 主题 选择主题 ( cloudlock-data-export-trigger)消息正文 {}(空 JSON 对象)点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次:
频率 Cron 表达式 使用场景 每隔 5 分钟 */5 * * * *高容量、低延迟 每隔 15 分钟 */15 * * * *搜索量中等 每小时 0 * * * *标准(推荐) 每 6 小时 0 */6 * * *量小、批处理 每天 0 0 * * *历史数据收集
测试调度器作业
- 在 Cloud Scheduler 控制台中,找到您的作业。
- 点击强制运行以手动触发。
- 等待几秒钟,然后前往 Cloud Run > 服务 > cloudlock-data-export > 日志。
- 验证函数是否已成功执行。
- 检查 GCS 存储分区,确认日志已写入。
检索 Google SecOps 服务账号
Google SecOps 使用唯一的服务账号从您的 GCS 存储分区中读取数据。您必须授予此服务账号对您的存储分区的访问权限。
获取服务账号电子邮件地址
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中,输入 Feed 的名称(例如
Cisco CloudLock logs)。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Cisco CloudLock 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址,以便在下一步中使用。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要对您的 GCS 存储分区具有 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色:选择 Storage Object Viewer。
- 点击保存。
在 Google SecOps 中配置 Feed 以注入 Cisco CloudLock 日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中,输入 Feed 的名称(例如
Cisco CloudLock logs)。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Cisco CloudLock 作为日志类型。
- 点击下一步。
为以下输入参数指定值:
存储分区网址:输入带有前缀路径的 GCS 存储分区 URI:
gs://cisco-cloudlock-logs/cloudlock/将
cisco-cloudlock-logs:您的 GCS 存储分区名称。cloudlock/:存储日志的可选前缀/文件夹路径(留空表示根目录)。
示例:
- 根存储分区:
gs://cisco-cloudlock-logs/ - 带前缀:
gs://cisco-cloudlock-logs/cloudlock/ - 使用子文件夹:
gs://cisco-cloudlock-logs/cloudlock/incidents/
- 根存储分区:
来源删除选项:根据您的偏好选择删除选项:
- 永不:永不删除转移后的任何文件(建议用于测试)。
- 删除已转移的文件:在成功转移后删除文件。
删除已转移的文件和空目录:在成功转移后删除文件和空目录。
文件存在时间上限:包含在过去指定天数内修改的文件。默认值为 180 天。
资产命名空间:资产命名空间。
注入标签:要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置,然后点击提交。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
| created_at | about.resource.attribute.labels.key | created_at 字段的值会分配给 labels 键。 |
| created_at | about.resource.attribute.labels.value | created_at 字段的值会分配给标签值。 |
| created_at | about.resource.attribute.creation_time | created_at 字段会解析为时间戳并进行映射。 |
| entity.id | target.asset.product_object_id | 实体.id 字段已重命名。 |
| entity.ip | target.ip | entity.ip 字段已合并到目标 IP 字段中。 |
| entity.mime_type | target.file.mime_type | 当 entity.origin_type 为“document”时,entity.mime_type 字段会重命名。 |
| entity.name | target.application | 当 entity.origin_type 为“app”时,entity.name 字段会被重命名。 |
| entity.name | target.file.full_path | 当 entity.origin_type 为“document”时,entity.name 字段会重命名。 |
| entity.origin_id | target.resource.product_object_id | entity.origin_id 字段已更名。 |
| entity.origin_type | target.resource.resource_subtype | entity.origin_type 字段已更名。 |
| entity.owner_email | target.user.email_addresses | 如果 entity.owner_email 字段与电子邮件正则表达式匹配,则会将其合并到目标用户电子邮件字段中。 |
| entity.owner_email | target.user.user_display_name | 如果 entity.owner_email 字段与电子邮件地址正则表达式不匹配,则会重命名该字段。 |
| entity.owner_name | target.user.user_display_name | 当 entity.owner_email 与电子邮件地址正则表达式匹配时,entity.owner_name 字段会被重命名。 |
| entity.vendor.name | target.platform_version | entity.vendor.name 字段已重命名。 |
| id | metadata.product_log_id | id 字段已更名。 |
| incident_status | metadata.product_event_type | 重命名了 incident_status 字段。 |
| metadata.event_timestamp | 值硬编码为“updated_at”。值派生自 updated_at 字段。updated_at 字段会被解析为时间戳并进行映射。 | |
| security_result.detection_fields.key | 如果严重程度为“ALERT”,且 incident_status 为“NEW”,则设置为“true”。转换为布尔值。 | |
| security_result.detection_fields.value | 如果严重程度为“ALERT”,且 incident_status 为“NEW”,则设置为“true”。转换为布尔值。 | |
| metadata.event_type | 值硬编码为“GENERIC_EVENT”。 | |
| metadata.product_name | 值硬编码为“CISCO_CLOUDLOCK_CASB”。 | |
| metadata.vendor_name | 值硬编码为“CloudLock”。 | |
| metadata.product_version | 值硬编码为“Cisco”。 | |
| security_result.alert_state | 如果严重程度为“ALERT”且 incident_status 不是“RESOLVED”或“DISMISSED”,则设置为“ALERTING”。如果严重程度为“ALERT”且 incident_status 为“RESOLVED”或“DISMISSED”,则设置为“NOT_ALERTING”。 | |
| security_result.detection_fields.key | 派生自 matches 数组,具体来说是每个匹配对象的键。 | |
| security_result.detection_fields.value | 派生自 matches 数组,具体来说是每个匹配对象的值。 | |
| security_result.rule_id | 派生自 policy.id。 | |
| security_result.rule_name | 派生自 policy.name。 | |
| security_result.severity | 如果严重程度为“INFO”,则设置为“INFORMATIONAL”。如果严重程度为“CRITICAL”,则设置为“CRITICAL”。派生自严重程度。 | |
| security_result.summary | 该值设置为“match count: ”与 match_count 值的串联。 | |
| target.resource.resource_type | 当 entity.origin_type 为“document”时,设置为“STORAGE_OBJECT”。 | |
| target.url | 当 entity.origin_type 为“document”时,派生自 entity.direct_url。 | |
| policy.id | security_result.rule_id | 已重命名 policy.id 字段。 |
| policy.name | security_result.rule_name | 重命名了 policy.name 字段。 |
| 和程度上减少 | security_result.severity_details | 重命名了严重程度字段。 |
| updated_at | about.resource.attribute.labels.key | updated_at 字段的值会分配给 labels 键。 |
| updated_at | about.resource.attribute.labels.value | 将 updated_at 字段的值分配给标签值。 |
| updated_at | about.resource.attribute.last_update_time | updated_at 字段会被解析为时间戳并进行映射。 |
需要更多帮助?获得社区成员和 Google SecOps 专业人士的解答。