收集 Cisco CloudLock CASB 日志

解析器版本:3.0

支持的平台:

本文档介绍了如何使用 Google Cloud Storage 将 Cisco CloudLock CASB 日志注入到 Google Security Operations。Cloud Run 函数会按计划轮询 Cisco CloudLock API,并将事件、活动和实体数据写入 GCS 存储桶,然后 Google SecOps 通过 GCS V2 Feed 提取这些数据。

Cisco CloudLock 是一款云原生云访问安全代理 (CASB),可提供对云应用的可见性和控制。它可帮助组织发现影子 IT、强制执行数据泄露防护政策、检测威胁,并确保 SaaS 应用符合相关法规。

准备工作

请确保您满足以下前提条件:

  • Google SecOps 实例
  • 已启用 Cloud Storage API 的 GCP 项目
  • 创建和管理 GCS 存储分区的权限
  • 管理 GCS 存储分区的 IAM 政策的权限
  • 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
  • 对 Cisco CloudLock 管理控制台的特权访问权限
  • Cisco CloudLock API 基本网址(由 Cisco CloudLock 支持团队提供)

获取 Cisco CloudLock API 凭据

如需开始使用,请与 Cisco CloudLock 支持团队联系,获取您的 CloudLock API 网址。API 基本网址特定于您的租户,且不对外公开。在 CloudLock 应用中生成访问令牌,方法是前往“设置”页面中的“身份验证和 API”标签页。

  1. 登录 Cisco CloudLock 管理控制台。
  2. 前往设置 > 身份验证和 API
  3. API 下,点击 Generate(生成)以创建访问令牌。
  4. 复制以下详细信息并将其保存在安全的位置:

    • API 访问令牌:点击“生成”后显示的不记名令牌。
    • API 基准网址:由 Cisco CloudLock 支持团队提供。格式为 https://{YourCloudlockAPIServer}

验证 API 访问权限

  • 在继续进行集成之前,请先测试您的凭据:

    CLOUDLOCK_API_TOKEN="your-api-token"
    CLOUDLOCK_API_BASE="https://your-cloudlock-api-server"
    
    curl -v -H "Authorization: Bearer ${CLOUDLOCK_API_TOKEN}" \
      "${CLOUDLOCK_API_BASE}/api/v2/incidents?limit=1"
    

如果响应成功,则返回 HTTP 200,并附带一个包含 items 数组的 JSON 对象。如果您收到 HTTP 401,请验证 API 令牌。如果您收到 HTTP 403 错误,请确认相应账号拥有所需的权限。

创建 Google Cloud Storage 存储桶

  1. 前往 Google Cloud 控制台
  2. 选择您的项目或创建新项目。
  3. 在导航菜单中,依次前往 Cloud Storage > 存储分区
  4. 点击创建存储分区
  5. 提供以下配置详细信息:

    设置
    为存储桶命名 输入一个全局唯一的名称(例如 cisco-cloudlock-logs
    位置类型 根据您的需求进行选择(区域级、双区域、多区域)
    位置 选择营业地点(例如 us-central1
    存储类别 标准(建议用于经常访问的日志)
    访问权限控制 均匀(推荐)
    保护工具 可选:启用对象版本控制或保留政策
  6. 点击创建

为 Cloud Run 函数创建服务账号

Cloud Run 函数需要一个服务账号,该账号具有写入 GCS 存储桶的权限,并且可以由 Pub/Sub 调用。

创建服务账号

  1. GCP 控制台中,依次前往 IAM 和管理 > 服务账号
  2. 点击创建服务账号
  3. 提供以下配置详细信息:
    • 服务账号名称:输入 cloudlock-data-export-sa
    • 服务账号说明:输入 Service account for Cloud Run function to collect Cisco CloudLock logs
  4. 点击创建并继续
  5. 向此服务账号授予对项目的访问权限部分:
    1. 点击选择角色
    2. 搜索并选择 Storage Object Admin
    3. 点击 + 添加其他角色
    4. 搜索并选择 Cloud Run Invoker
    5. 点击 + 添加其他角色
    6. 搜索并选择 Cloud Functions Invoker
  6. 点击继续
  7. 点击完成

必须拥有这些角色,才能:

  • Storage Object Admin:将日志写入 GCS 存储桶并管理状态文件
  • Cloud Run Invoker:允许 Pub/Sub 调用函数
  • Cloud Functions Invoker:允许调用函数

授予对 GCS 存储桶的 IAM 权限

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

  1. 前往 Cloud Storage > 存储分区
  2. 点击您的存储桶名称。
  3. 前往权限标签页。
  4. 点击授予访问权限
  5. 提供以下配置详细信息:
    • 添加主账号:输入服务账号电子邮件地址(例如 cloudlock-data-export-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:输入 cloudlock-data-export-trigger
    • 将其他设置保留为默认值。
  4. 点击创建

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发,以从 Cisco CloudLock API 中提取日志并将其写入 GCS。

  1. GCP 控制台中,前往 Cloud Run
  2. 点击创建服务
  3. 选择函数(使用内嵌编辑器创建函数)。
  4. 配置部分中,提供以下配置详细信息:

    设置
    Service 名称 cloudlock-data-export
    区域 选择与您的 GCS 存储桶匹配的区域(例如 us-central1
    运行时 选择 Python 3.12 或更高版本
  5. 触发器(可选)部分中:

    1. 点击 + 添加触发器
    2. 选择 Cloud Pub/Sub
    3. 选择 Cloud Pub/Sub 主题中,选择相应主题 (cloudlock-data-export-trigger)。
    4. 点击保存
  6. 身份验证部分中:

    1. 选择需要进行身份验证
    2. 检查 Identity and Access Management (IAM)
  7. 向下滚动并展开容器、网络、安全性

  8. 前往安全性标签页:

    • 服务账号:选择服务账号 (cloudlock-data-export-sa)。
  9. 前往容器标签页:

    1. 点击变量和密钥
    2. 为每个环境变量点击+ 添加变量
    变量名称 示例值 说明
    GCS_BUCKET cisco-cloudlock-logs GCS 存储桶名称
    GCS_PREFIX cloudlock/ 日志文件的前缀
    STATE_KEY cloudlock/state.json 状态文件路径
    CLOUDLOCK_API_TOKEN your-api-token API 不记名令牌
    CLOUDLOCK_API_BASE https://your-cloudlock-api-server CloudLock 支持团队提供的 API 基本网址
  10. 变量和密钥标签页中,向下滚动到请求

    • 请求超时:输入 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, 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 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."""
          backoff = 1.0
          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 - CloudLock enforces per-endpoint rate limits
                      retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                      print(f'Rate limited (429), waiting {retry_after} seconds')
                      time.sleep(retry_after)
                      backoff = min(backoff * 2, 60.0)
                  else:
                      print(f'API request failed with status {response.status}: {response.data.decode("utf-8")}')
                      if response.status in (401, 403):
                          return None
              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.
      
          Endpoint: GET /api/v2/incidents
          Pagination: limit and offset query parameters.
          Response: JSON with 'items' array, 'total' count, and 'results' count.
          """
          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.
      
          Endpoint: GET /api/v2/activities
          Pagination: limit and offset query parameters.
          Note: Client-side time filtering is applied since the API may not support time range parameters.
          """
          url = f"{api_base}/api/v2/activities"
      
          limit = 1000
          offset = start_offset
          all_data = []
          newest_time = None
      
          try:
              while True:
                  # Build URL with 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)
                  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.
      
          Endpoint: GET /api/v2/entities
          Note: This endpoint requires the Entity Cache feature. If not enabled,
          use the incident entities endpoint (/api/v2/incidents/{id}/entities) 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
      
  3. 点击部署以保存并部署该函数。

  4. 等待部署完成(2-3 分钟)。

创建 Cloud Scheduler 作业

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

  1. GCP Console 中,前往 Cloud Scheduler
  2. 点击创建作业
  3. 提供以下配置详细信息:

    设置
    名称 cloudlock-data-export-hourly
    区域 选择与 Cloud Run 函数相同的区域
    频率 0 * * * *(每小时一次,整点时)
    时区 选择时区(建议选择 UTC)
    目标类型 Pub/Sub
    主题 选择主题 (cloudlock-data-export-trigger)
    消息正文 {}(空 JSON 对象)
  4. 点击创建

时间表频率选项

根据日志量和延迟时间要求选择频次:

频率 Cron 表达式 使用场景
每隔 5 分钟 */5 * * * * 大批量、低延迟
每隔 15 分钟 */15 * * * * 搜索量中等
每小时 0 * * * * 标准(推荐)
每 6 小时 0 */6 * * * 低成交量、批处理
每天 0 0 * * * 历史数据收集

测试集成

  1. Cloud Scheduler 控制台中,找到您的作业。
  2. 点击强制运行以手动触发作业。
  3. 等待几秒钟。
  4. 前往 Cloud Run > 服务
  5. 点击函数名称 (cloudlock-data-export)。
  6. 点击日志标签页。
  7. 验证函数是否已成功执行。查找:

    Processing logs with state: {}
    Fetching incidents with offset: 0
    Fetched X incidents (total available: Y)
    Uploaded X incidents to GCS
    CloudLock data export completed successfully
    
  8. 前往 Cloud Storage > 存储分区

  9. 点击相应存储桶的名称 (cisco-cloudlock-logs)。

  10. 前往 cloudlock/ 前缀文件夹。

  11. 验证是否在 incidents/activities/entities/ 子文件夹下创建了新的 .jsonl 文件。

如果您在日志中看到错误,请执行以下操作:

  • HTTP 401:检查环境变量中的 API 令牌
  • HTTP 403:验证 CloudLock 账号是否具有所需权限
  • HTTP 429:速率限制 - 函数将自动重试并进行退避
  • 缺少环境变量:检查是否已设置所有必需的变量
  • 空实体:实体端点需要实体缓存功能 - 请与 CloudLock 支持团队联系以启用该功能

在 Google SecOps 中配置 Feed 以注入 Cisco CloudLock 日志

  1. 依次前往 SIEM 设置 > Feed
  2. 点击添加新 Feed
  3. 点击配置单个 Feed
  4. Feed 名称字段中,输入 Feed 的名称(例如 Cisco CloudLock logs)。
  5. 选择 Google Cloud Storage V2 作为来源类型
  6. 选择 Cisco CloudLock 作为日志类型
  7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 复制此电子邮件地址,以便在下一步中使用。

  9. 点击下一步

  10. 为以下输入参数指定值:

    • 存储桶网址:输入带有前缀路径的 GCS 存储桶 URI:

      gs://cisco-cloudlock-logs/cloudlock/
      
      • 替换:
        • cisco-cloudlock-logs:您的 GCS 存储桶名称。
        • cloudlock/:存储日志的可选前缀/文件夹路径(留空表示根目录)。
    • 来源删除选项:根据您的偏好选择删除选项:

      • 永不:转移后永不删除任何文件(建议用于测试)。
      • 删除已转移的文件:在成功转移后删除文件。
      • 删除已转移的文件和空目录:成功转移后删除文件和空目录。
    • 文件存在时间上限:包含在过去指定天数内修改的文件(默认值为 180 天)。

    • 资产命名空间资产命名空间

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

  11. 点击下一步

  12. 最终确定界面中查看新的 Feed 配置,然后点击提交

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

Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。

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

UDM 映射表

日志字段 UDM 映射 逻辑
created_at about.resource.attribute.creation_time 原始日志中的 created_at 值会被解析为时间戳并映射到 about.resource.attribute.creation_time 字段。
created_at about.resource.attribute.labels.key created_at 的值会分配给标签键。
created_at about.resource.attribute.labels.value created_at 的值会分配给标签值。
entity.id target.asset.product_object_id 原始日志中的 entity.id 值会映射到 target.asset.product_object_id 字段。
entity.ip target.ip 原始日志中的 entity.ip 值会合并到 target.ip 字段中。
entity.mime_type target.file.mime_type entity.origin_type 为“document”时,原始日志中的 entity.mime_type 值会映射到 target.file.mime_type 字段。
entity.name target.application entity.origin_type 为“app”时,原始日志中的 entity.name 值会映射到 target.application 字段。
entity.name target.file.full_path entity.origin_type 为“document”时,原始日志中的 entity.name 值会映射到 target.file.full_path 字段。
entity.origin_id target.resource.product_object_id 原始日志中的 entity.origin_id 值会映射到 target.resource.product_object_id 字段。
entity.origin_type target.resource.resource_subtype 原始日志中的 entity.origin_type 值会映射到 target.resource.resource_subtype 字段。
entity.owner_email target.user.email_addresses 如果原始日志中的 entity.owner_email 值与电子邮件正则表达式匹配,则会将其合并到 target.user.email_addresses 字段中。
entity.owner_email target.user.user_display_name 如果原始日志中的 entity.owner_email 值与电子邮件正则表达式不匹配,则会将其映射到 target.user.user_display_name 字段。
entity.owner_name target.user.user_display_name entity.owner_email 与电子邮件正则表达式匹配时,原始日志中的 entity.owner_name 值会映射到 target.user.user_display_name 字段。
entity.vendor.name target.platform_version 原始日志中的 entity.vendor.name 值会映射到 target.platform_version 字段。
id metadata.product_log_id 原始日志中的 id 值会映射到 metadata.product_log_id 字段。
incident_status metadata.product_event_type 原始日志中的 incident_status 值会映射到 metadata.product_event_type 字段。
policy.id security_result.rule_id 原始日志中的 policy.id 值会映射到 security_result.rule_id 字段。
policy.name security_result.rule_name 原始日志中的 policy.name 值会映射到 security_result.rule_name 字段。
和程度上减少 security_result.severity_details 原始日志中的 severity 值会映射到 security_result.severity_details 字段。
updated_at about.resource.attribute.labels.key updated_at 的值会分配给标签键。
updated_at about.resource.attribute.labels.value updated_at 的值会分配给标签值。
updated_at about.resource.attribute.last_update_time 原始日志中的 updated_at 值会被解析为时间戳并映射到 about.resource.attribute.last_update_time 字段。
不适用 metadata.event_timestamp 派生自 updated_at 字段。该值会被解析为时间戳并进行映射。
不适用 metadata.event_type 设置为“GENERIC_EVENT”。
不适用 metadata.product_name 设置为“CISCO_CLOUDLOCK_CASB”。
不适用 metadata.product_version 设置为“Cisco”。
不适用 metadata.vendor_name 设置为“CloudLock”。
不适用 security_result.alert_state 如果 severity 为“ALERT”,且 incident_status 不为“RESOLVED”或“DISMISSED”,则设置为“ALERTING”。如果 severity 为“ALERT”,且 incident_status 为“RESOLVED”或“DISMISSED”,则设置为“NOT_ALERTING”。
不适用 security_result.detection_fields.key 派生自 matches 数组,具体来说是每个匹配对象的键。如果 severity 为“ALERT”且 incident_status 为“NEW”,则也设置为“true”。
不适用 security_result.detection_fields.value 派生自 matches 数组,具体来说是每个匹配对象的值。如果 severity 为“ALERT”且 incident_status 为“NEW”,则也设置为“true”。
不适用 security_result.severity 如果 severity 为“INFO”,则设置为“INFORMATIONAL”。如果 severity 为“CRITICAL”,则设置为“CRITICAL”。派生自 severity 字段。
不适用 security_result.summary 设置为“匹配次数:”与 match_count 的值串联。
不适用 target.resource.resource_type entity.origin_type 为“document”时,设置为“STORAGE_OBJECT”。
不适用 target.url entity.origin_type 为“document”时,派生自 entity.direct_url

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