收集 Censys 日志

支持的平台：

本文档介绍了如何使用 Google Cloud Storage V2 将 Censys 日志提取到 Google Security Operations。

Censys 通过其 API 提供全面的攻击面管理和互联网情报。通过此集成，您可以从 Censys ASM 收集主机发现事件、风险事件和资产变更，并将它们转发到 Google SecOps 以进行分析和监控。

准备工作

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

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

收集 Censys API 凭据

  1. 前往 app.censys.io，登录 Censys ASM 控制台。
  2. 前往页面顶部的集成。
  3. 复制并保存您的 API 密钥和组织 ID。
  4. 请注意 API 基准网址：https://api.platform.censys.io

创建 Google Cloud Storage 存储分区

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

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

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

  6. 点击创建。

为 Cloud Run 函数创建服务账号

Cloud Run 函数需要一个有权写入 GCS 存储分区的服务账号。

创建服务账号

  1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
  2. 点击创建服务账号。
  3. 提供以下配置详细信息：
    • 服务账号名称：输入 censys-data-collector-sa。
    • 服务账号说明：输入 Service account for Cloud Run function to collect Censys 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. 提供以下配置详细信息：
    • 添加主账号：输入服务账号电子邮件地址（例如 censys-data-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。
    • 分配角色：选择 Storage Object Admin。
  6. 点击保存。

创建发布/订阅主题

创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。

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

创建 Cloud Run 函数以收集日志

Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发，用于从 Censys ASM API 中提取日志并将其写入 GCS。

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

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

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

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

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

  6. 在身份验证部分中：

    • 选择需要进行身份验证。
    • 检查 Identity and Access Management (IAM)。
  1. 向下滚动并展开容器、网络、安全性。
  2. 前往安全性标签页：
    • 服务账号：选择服务账号 (censys-data-collector-sa)。

  3. 前往容器标签页：

    1. 点击变量和密钥。

    2. 为每个环境变量点击 + 添加变量：

      变量名称 示例值
      GCS_BUCKET censys-logs
      GCS_PREFIX censys/
      STATE_KEY censys/state.json
      CENSYS_API_KEY your-censys-api-key
      CENSYS_ORG_ID your-organization-id
      API_BASE https://api.platform.censys.io

  4. 在变量和密钥标签页中，向下滚动到请求：

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

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

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

  6. 滚动到执行环境：

    • 选择默认（推荐）。

  7. 在修订版本扩缩部分中：

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

  8. 点击创建。

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

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

添加函数代码

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

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

    • 第一个文件：main.py:：
    import functions_framework
from google.cloud import storage
import json
import urllib3
import gzip
import os
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Any, Optional
from urllib.parse import urlencode

# 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 Censys ASM 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', 'censys/')
    state_key = os.environ.get('STATE_KEY', 'censys/state.json')
    censys_api_key = os.environ.get('CENSYS_API_KEY')
    censys_org_id = os.environ.get('CENSYS_ORG_ID')
    api_base = os.environ.get('API_BASE', 'https://api.platform.censys.io')

    if not all([bucket_name, censys_api_key, censys_org_id]):
        print('Error: Missing required environment variables')
        return

    try:
        collector = CensysCollector(
            bucket_name=bucket_name,
            prefix=prefix,
            state_key=state_key,
            api_key=censys_api_key,
            org_id=censys_org_id,
            api_base=api_base
        )

        # Get last collection time
        last_collection_time = collector.get_last_collection_time()
        current_time = datetime.now(timezone.utc)

        print(f'Collecting events since {last_collection_time}')

        # Collect different types of events
        logbook_events = collector.collect_logbook_events()
        risk_events = collector.collect_risks_events()

        # Save events to GCS
        collector.save_events_to_gcs(logbook_events, 'logbook')
        collector.save_events_to_gcs(risk_events, 'risks')

        # Update state
        collector.save_collection_time(current_time)

        print(f'Successfully processed {len(logbook_events)} logbook events and {len(risk_events)} risk events')

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

class CensysCollector:
    def __init__(self, bucket_name: str, prefix: str, state_key: str, 
                 api_key: str, org_id: str, api_base: str):
        self.bucket_name = bucket_name
        self.prefix = prefix
        self.state_key = state_key
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'X-Organization-ID': org_id,
            'Content-Type': 'application/json'
        }
        self.api_base = api_base
        self.bucket = storage_client.bucket(bucket_name)

    def get_last_collection_time(self) -> Optional[datetime]:
        """Get the last collection timestamp from GCS state file."""
        try:
            blob = self.bucket.blob(self.state_key)
            if blob.exists():
                state_data = blob.download_as_text()
                state = json.loads(state_data)
                return datetime.fromisoformat(state.get('last_collection_time', '2024-01-01T00:00:00Z'))
        except Exception as e:
            print(f'No state file found or error reading state: {e}')
        return datetime.now(timezone.utc) - timedelta(hours=1)

    def save_collection_time(self, collection_time: datetime):
        """Save the current collection timestamp to GCS state file."""
        state = {'last_collection_time': collection_time.strftime('%Y-%m-%dT%H:%M:%SZ')}
        blob = self.bucket.blob(self.state_key)
        blob.upload_from_string(
            json.dumps(state),
            content_type='application/json'
        )

    def collect_logbook_events(self, cursor: str = None) -> List[Dict[str, Any]]:
        """Collect logbook events from Censys ASM API using cursor-based pagination."""
        events = []
        url = f"{self.api_base}/v3/logbook"

        params = {}
        if cursor:
            params['cursor'] = cursor

        try:
            query_string = urlencode(params) if params else ''
            full_url = f"{url}?{query_string}" if query_string else url

            response = http.request('GET', full_url, headers=self.headers)

            # Handle rate limiting with exponential backoff
            if response.status == 429:
                retry_after = int(response.headers.get('Retry-After', '60'))
                print(f'Rate limited (429). Retrying after {retry_after}s...')
                import time
                time.sleep(retry_after)
                return self.collect_logbook_events(cursor)

            if response.status != 200:
                print(f'API request failed with status {response.status}: {response.data}')
                return []

            data = json.loads(response.data.decode('utf-8'))
            events.extend(data.get('logbook_entries', []))

            # Handle cursor-based pagination
            next_cursor = data.get('next_cursor')
            if next_cursor:
                events.extend(self.collect_logbook_events(next_cursor))

            print(f'Collected {len(events)} logbook events')
            return events

        except Exception as e:
            print(f'Error collecting logbook events: {e}')
            return []

    def collect_risks_events(self) -> List[Dict[str, Any]]:
        """Collect risk events from Censys ASM API."""
        events = []
        url = f"{self.api_base}/v3/risks"

        try:
            response = http.request('GET', url, headers=self.headers)

            # Handle rate limiting with exponential backoff
            if response.status == 429:
                retry_after = int(response.headers.get('Retry-After', '60'))
                print(f'Rate limited (429). Retrying after {retry_after}s...')
                import time
                time.sleep(retry_after)
                return self.collect_risks_events()

            if response.status != 200:
                print(f'API request failed with status {response.status}: {response.data}')
                return []

            data = json.loads(response.data.decode('utf-8'))
            events.extend(data.get('risks', []))

            print(f'Collected {len(events)} risk events')
            return events

        except Exception as e:
            print(f'Error collecting risk events: {e}')
            return []

    def save_events_to_gcs(self, events: List[Dict[str, Any]], event_type: str):
        """Save events to GCS in compressed NDJSON format."""
        if not events:
            return

        timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
        filename = f"{self.prefix}{event_type}_{timestamp}.json.gz"

        try:
            # Convert events to newline-delimited JSON
            ndjson_content = '\n'.join(json.dumps(event, separators=(',', ':')) for event in events)

            # Compress with gzip
            gz_bytes = gzip.compress(ndjson_content.encode('utf-8'))

            blob = self.bucket.blob(filename)
            blob.upload_from_string(
                gz_bytes,
                content_type='application/gzip'
            )

            print(f'Saved {len(events)} {event_type} events to {filename}')

        except Exception as e:
            print(f'Error saving {event_type} events to GCS: {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. 提供以下配置详细信息：

    设置 值
    名称 censys-data-collector-hourly
    区域 选择与 Cloud Run 函数相同的区域
    频率 0 * * * *（每小时一次，在整点时）
    时区 选择时区（建议选择世界协调时间 [UTC]）
    目标类型 Pub/Sub
    主题 选择 Pub/Sub 主题 (censys-data-trigger)
    消息正文 {}（空 JSON 对象）

  4. 点击创建。

时间表频率选项

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

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

测试集成

  1. 在 Cloud Scheduler 控制台中，找到您的作业。
  2. 点击强制运行以手动触发作业。
  3. 等待几秒钟。
  4. 前往 Cloud Run > 服务。
  5. 点击函数名称 (censys-data-collector)。
  6. 点击日志标签页。

  7. 验证函数是否已成功执行。查找以下项：

    Collecting events since YYYY-MM-DDTHH:MM:SS+00:00
Collected X logbook events
Collected X risk events
Saved X logbook events to censys/logbook_YYYYMMDD_HHMMSS.json.gz
Saved X risks events to censys/risks_YYYYMMDD_HHMMSS.json.gz
Successfully processed X logbook events and X risk events

  8. 前往 Cloud Storage > 存储分区。

  9. 点击您的存储分区名称。

  10. 前往前缀文件夹 (censys/)。

  11. 验证是否已创建具有当前时间戳的新 .json.gz 文件。

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

  • HTTP 401：检查环境变量中的 API 凭据
  • HTTP 403：验证账号是否具有所需权限
  • HTTP 429：速率限制 - 函数将自动重试并进行退避
  • 缺少环境变量：检查是否已设置所有必需的变量

检索 Google SecOps 服务账号

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

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

  1. 依次前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 点击配置单个 Feed。
  4. 在Feed 名称字段中，输入 Feed 的名称（例如 Censys logs）。
  5. 选择 Google Cloud Storage V2 作为来源类型。
  6. 选择 CENSYS 作为日志类型。
  7. 点击获取服务账号。

  8. 系统会显示一个唯一的服务账号电子邮件地址，例如：

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

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

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

Google SecOps 服务账号需要对您的 GCS 存储分区具有 Storage Object Viewer 角色。

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

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

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

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

    • 存储分区网址：输入带有前缀路径的 GCS 存储分区 URI：

      gs://censys-logs/censys/

      • 将

        • censys-logs：您的 GCS 存储分区名称。
        • censys/：存储日志的可选前缀/文件夹路径（留空表示根目录）。

      • 示例：

        • 根存储分区：gs://censys-logs/
        • 带前缀：gs://censys-logs/censys/

    • 来源删除选项：根据您的偏好选择删除选项：

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

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

    • 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。

    • 资产命名空间：资产命名空间。

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

  9. 点击下一步。

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

UDM 映射表

日志字段 UDM 映射 逻辑
assetId read_only_udm.principal.asset.hostname 如果 assetId 字段不是 IP 地址，则会映射到 principal.asset.hostname。
assetId read_only_udm.principal.asset.ip 如果 assetId 字段是 IP 地址，则会映射到 principal.asset.ip。
assetId read_only_udm.principal.hostname 如果 assetId 字段不是 IP 地址，则会映射到 principal.hostname。
assetId read_only_udm.principal.ip 如果 assetId 字段是 IP 地址，则会映射到 principal.ip。
associatedAt read_only_udm.security_result.detection_fields.value associatedAt 字段映射到 security_result.detection_fields.value。
autonomousSystem.asn read_only_udm.additional.fields.value.string_value autonomousSystem.asn 字段会转换为字符串，并映射到 additional.fields.value.string_value，键为“autonomousSystem_asn”。
autonomousSystem.bgpPrefix read_only_udm.additional.fields.value.string_value autonomousSystem.bgpPrefix 字段映射到 additional.fields.value.string_value，键为“autonomousSystem_bgpPrefix”。
横幅 read_only_udm.principal.resource.attribute.labels.value 横幅字段映射到键为“banner”的 principal.resource.attribute.labels.value。
云 read_only_udm.metadata.vendor_name 云字段映射到 metadata.vendor_name。
comments.refUrl read_only_udm.network.http.referral_url comments.refUrl 字段映射到 network.http.referral_url。
data.cve read_only_udm.additional.fields.value.string_value data.cve 字段映射到键为“data_cve”的 additional.fields.value.string_value。
data.cvss read_only_udm.additional.fields.value.string_value data.cvss 字段映射到键为“data_cvss”的 additional.fields.value.string_value。
data.ipAddress read_only_udm.principal.asset.ip 如果 data.ipAddress 字段不等于 assetId 字段，则会映射到 principal.asset.ip。
data.ipAddress read_only_udm.principal.ip 如果 data.ipAddress 字段不等于 assetId 字段，则会映射到 principal.ip。
data.location.city read_only_udm.principal.location.city 如果 location.city 字段为空，则将 data.location.city 字段映射到 principal.location.city。
data.location.countryCode read_only_udm.principal.location.country_or_region 如果 location.country 字段为空，则 data.location.countryCode 字段会映射到 principal.location.country_or_region。
data.location.latitude read_only_udm.principal.location.region_coordinates.latitude 如果 location.coordinates.latitude 和 location.geoCoordinates.latitude 字段为空，则将 data.location.latitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
data.location.longitude read_only_udm.principal.location.region_coordinates.longitude 如果 location.coordinates.longitude 和 location.geoCoordinates.longitude 字段为空，则将 data.location.longitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
data.location.province read_only_udm.principal.location.state 如果 location.province 字段为空，则 data.location.province 字段会映射到 principal.location.state。
data.mailServers read_only_udm.additional.fields.value.list_value.values.string_value data.mailServers 数组中的每个元素都会映射到单独的 additional.fields 条目，其中键为“Mail Servers”，value.list_value.values.string_value 设置为相应元素的值。
data.names.forwardDns[].name read_only_udm.network.dns.questions.name data.names.forwardDns 数组中的每个元素都会映射到单独的 network.dns.questions 条目，并将 name 字段设置为相应元素的 name 字段。
data.nameServers read_only_udm.additional.fields.value.list_value.values.string_value data.nameServers 数组中的每个元素都会映射到单独的 additional.fields 条目，其中键为“Name nameServers”，value.list_value.values.string_value 设置为相应元素的值。
data.protocols[].transportProtocol read_only_udm.network.ip_protocol 如果 data.protocols[].transportProtocol 字段是 TCP、EIGRP、ESP、ETHERIP、GRE、ICMP、IGMP、IP6IN4、PIM、UDP 或 VRRP 之一，则会映射到 network.ip_protocol。
data.protocols[].transportProtocol read_only_udm.principal.resource.attribute.labels.value data.protocols[].transportProtocol 字段映射到 principal.resource.attribute.labels.value，键为“data_protocols {index}”。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.network.http.user_agent 如果 http.request.headers[].key 字段为“User-Agent”，则相应的 http.request.headers[].value.headers.0 字段会映射到 network.http.user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.network.http.parsed_user_agent 如果 http.request.headers[].key 字段为“User-Agent”，则相应的 http.request.headers[].value.headers.0 字段会被解析为用户代理字符串并映射到 network.http.parsed_user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value 对于 http.request.headers 数组中的每个元素，key 字段会映射到 principal.resource.attribute.labels.key，而 value.headers.0 字段会映射到 principal.resource.attribute.labels.value。
http.request.uri read_only_udm.principal.asset.hostname 提取 http.request.uri 字段的主机名部分，并将其映射到 principal.asset.hostname。
http.request.uri read_only_udm.principal.hostname 提取 http.request.uri 字段的主机名部分，并将其映射到 principal.hostname。
http.response.body read_only_udm.principal.resource.attribute.labels.value http.response.body 字段映射到 principal.resource.attribute.labels.value，键为“http_response_body”。
http.response.headers[].key, http.response.headers[].value.headers.0 read_only_udm.target.hostname 如果 http.response.headers[].key 字段为“Server”，则相应的 http.response.headers[].value.headers.0 字段会映射到 target.hostname。
http.response.headers[].key, http.response.headers[].value.headers.0 read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value 对于 http.response.headers 数组中的每个元素，key 字段会映射到 principal.resource.attribute.labels.key，而 value.headers.0 字段会映射到 principal.resource.attribute.labels.value。
http.response.statusCode read_only_udm.network.http.response_code http.response.statusCode 字段会转换为整数并映射到 network.http.response_code。
ip read_only_udm.target.asset.ip ip 字段已映射到 target.asset.ip。
ip read_only_udm.target.ip ip 字段已映射到 target.ip。
isSeed read_only_udm.additional.fields.value.string_value isSeed 字段会转换为字符串，并映射到键为“isSeed”的 additional.fields.value.string_value。
location.city read_only_udm.principal.location.city location.city 字段会映射到 principal.location.city。
location.continent read_only_udm.additional.fields.value.string_value location.continent 字段映射到键为“location_continent”的 additional.fields.value.string_value。
location.coordinates.latitude read_only_udm.principal.location.region_coordinates.latitude location.coordinates.latitude 字段会转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
location.coordinates.longitude read_only_udm.principal.location.region_coordinates.longitude location.coordinates.longitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
location.country read_only_udm.principal.location.country_or_region location.country 字段会映射到 principal.location.country_or_region。
location.geoCoordinates.latitude read_only_udm.principal.location.region_coordinates.latitude 如果 location.coordinates.latitude 字段为空，则将 location.geoCoordinates.latitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
location.geoCoordinates.longitude read_only_udm.principal.location.region_coordinates.longitude 如果 location.coordinates.longitude 字段为空，则将 location.geoCoordinates.longitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
location.postalCode read_only_udm.additional.fields.value.string_value location.postalCode 字段映射到键为“邮政编码”的 additional.fields.value.string_value。
location.province read_only_udm.principal.location.state location.province 字段映射到 principal.location.state。
操作 read_only_udm.security_result.action_details 操作字段映射到 security_result.action_details。
perspectiveId read_only_udm.principal.group.product_object_id perspectiveId 字段映射到 principal.group.product_object_id。
端口 read_only_udm.principal.port 端口字段会转换为整数并映射到 principal.port。
risks[].severity、risks[].title read_only_udm.security_result.category_details 将 risks[].severity 字段与 risks[].title 字段串联，并映射到 security_result.category_details。
serviceName read_only_udm.network.application_protocol 如果 serviceName 字段为“HTTP”或“HTTPS”，则会映射到 network.application_protocol。
sourceIp read_only_udm.principal.asset.ip sourceIp 字段映射到 principal.asset.ip。
sourceIp read_only_udm.principal.ip sourceIp 字段映射到 principal.ip。
时间戳 read_only_udm.metadata.event_timestamp 时间戳字段会被解析为时间戳，并映射到 metadata.event_timestamp。
transportFingerprint.id read_only_udm.metadata.product_log_id transportFingerprint.id 字段转换为字符串并映射到 metadata.product_log_id。
transportFingerprint.raw read_only_udm.additional.fields.value.string_value transportFingerprint.raw 字段映射到键为“transportFingerprint_raw”的 additional.fields.value.string_value。
类型 read_only_udm.metadata.product_event_type type 字段会映射到 metadata.product_event_type。
- read_only_udm.metadata.product_name 值“CENSYS_ASM”已分配给 metadata.product_name。
- read_only_udm.metadata.vendor_name 值“CENSYS”已分配给 metadata.vendor_name。
- read_only_udm.metadata.event_type 事件类型根据特定字段的存在情况来确定：如果 has_princ_machine_id 和 has_target_machine 为 true 且 has_network_flow 为 false，则为 NETWORK_CONNECTION；如果 has_network_flow 为 true，则为 NETWORK_DNS；如果 has_princ_machine_id 为 true，则为 STATUS_UPDATE；否则为 GENERIC_EVENT。

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