收集 Dataminr Alerts 日志

支持的平台：

Google SecOps SIEM

注意： 此功能受《Google Security Operations 服务专用条款》的正式发布前产品条款约束。我们对正式发布前功能的支持可能有限，并且对正式发布前功能的更改与其他正式发布前版本可能不兼容。如需了解详情，请参阅 Google SecOps 技术支持服务准则和 Google SecOps 服务专用条款。

本文档介绍了如何使用 Google Cloud Storage V2、Cloud Run 函数和 Cloud Scheduler 将 Dataminr Alerts 日志注入到 Google Security Operations 中。

Dataminr Pulse 可从 50 多万个全球公共数据源（包括深层网络和暗网）中提取由 AI 提供支持的实时情报。该平台可针对影响贵组织和第三方的网络威胁、漏洞、勒索软件攻击、数据泄露和数字风险提供预警。Dataminr Pulse API 使用 OAuth 2.0 客户端凭据身份验证和基于光标的分页来检索提醒。

准备工作

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

• Google SecOps 实例
• 已启用以下 API 的 Google Cloud 项目：
  • Cloud Storage API
  • Cloud Run Functions API
  • Cloud Scheduler API
  • Cloud Pub/Sub API
• 创建和管理 GCS 存储分区、Cloud Run 函数、Pub/Sub 主题和 Cloud Scheduler 作业的权限
• 管理 GCS 存储分区的 IAM 政策的权限
• 已启用 API 访问权限的有效 Dataminr Pulse 账号
• Dataminr Pulse API 凭据（客户端 ID 和客户端密钥）
• 您的 Dataminr 账号中至少配置了一个 Dataminr Pulse 提醒列表

创建 Google Cloud Storage 存储分区

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

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

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

6. 点击创建。

收集 Dataminr 凭据

如需让 Cloud Run 函数能够检索提醒数据，您需要从 Dataminr 客户代表处获取具有 OAuth 2.0 客户端凭据身份验证的 API 凭据。

获取 API 凭据

1. 请与您的 Dataminr 客户代表或支持团队联系，申请 API 访问权限。
2. 请提供以下信息：
   • 您的组织名称
   • 使用情形：与 Google Chronicle SIEM 集成
   • 所需访问权限：Dataminr Pulse API for Cyber Risk

3. Dataminr 会提供 API 凭据，并向您提供以下信息：

   • 客户端 ID：您唯一的 OAuth 2.0 客户端标识符
   • 客户端密钥：您的 OAuth 2.0 客户端密钥
   重要提示： 请妥善存储客户端密钥。如果丢失，您必须向 Dataminr 申请新凭据。

验证 API 凭据

• 如需验证凭据是否有效，请运行以下命令：

  curl -X POST https://gateway.dataminr.com/auth/2/token \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET&grant_type=api_key"
  

  成功的响应会返回一个包含 access_token 字段的 JSON 对象：

  {
    "access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI...",
    "token_type": "Bearer",
    "expire": 3600
  }
  

收集提醒列表 ID

1. 登录 https://app.dataminr.com 上的 Dataminr Pulse Web 应用。
2. 前往您配置的提醒列表（关注列表）。

3. 记下要注入到 Google SecOps 中的提醒列表的 ID。

   注意： 配置 Cloud Run 函数环境变量时，您可以采用英文逗号分隔的字符串形式指定多个提醒列表 ID。

为 Cloud Run 函数创建服务账号

1. 在 Google Cloud 控制台中，依次前往 IAM 和管理 > 服务账号。
2. 点击创建服务账号。
3. 提供以下配置详细信息：
   • 服务账号名称：输入 dataminr-alert-collector
   • 服务账号说明：输入 Service account for Dataminr Alerts Cloud Run function to write alert data to GCS
4. 点击创建并继续。
5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
   1. 点击选择角色，然后搜索并选择 Storage Object Admin。
   2. 点击添加其他角色，然后搜索并选择 Cloud Run Invoker。
6. 点击继续。
7. 点击完成。

授予对 GCS 存储桶的 IAM 权限

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

创建 Pub/Sub 主题

当 Cloud Scheduler 发布消息时，Pub/Sub 主题会触发 Cloud Run 函数。

1. 在 Google Cloud 控制台中，前往 Pub/Sub > 主题。
2. 点击创建主题。
3. 提供以下配置详细信息：
   • 主题 ID：输入 dataminr-alert-trigger
   • 添加默认订阅：保持选中状态
4. 点击创建。

创建 Cloud Run 函数

1. 在 Google Cloud 控制台中，前往 Cloud Run 函数。
2. 点击创建函数。

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

   设置	值
   环境	第 2 代
   函数名称	dataminr-alert-collector
   区域	选择与您的 GCS 存储桶相同的区域
   触发器类型	Cloud Pub/Sub
   发布/订阅主题	dataminr-alert-trigger
   分配的内存	512 MiB
   超时	540 秒
   运行时服务账号	dataminr-alert-collector

4. 点击下一步。

5. 将运行时设置为 Python 3.12。

6. 将入口点设置为 main。

7. 在 requirements.txt 文件中，添加以下依赖项：

   functions-framework==3.*
   google-cloud-storage==2.*
   requests==2.*
   

8. 在 main.py 文件中，粘贴以下代码：

   import functions_framework
   import json
   import os
   import logging
   import time
   from datetime import datetime, timedelta, timezone
   from google.cloud import storage
   import requests
   
   logger = logging.getLogger(__name__)
   logger.setLevel(logging.INFO)
   
   storage_client = storage.Client()
   
   TOKEN_URL = "https://gateway.dataminr.com/auth/2/token"
   ALERTS_URL = "https://gateway.dataminr.com/api/3/alerts"
   
   def _get_access_token(client_id: str, client_secret: str) -> str:
       """Obtain an OAuth 2.0 access token from Dataminr."""
       payload = {
           "client_id": client_id,
           "client_secret": client_secret,
           "grant_type": "api_key",
       }
       headers = {"Content-Type": "application/x-www-form-urlencoded"}
       resp = requests.post(TOKEN_URL, data=payload, headers=headers, timeout=30)
       resp.raise_for_status()
       token_data = resp.json()
       access_token = token_data.get("access_token")
       if not access_token:
           raise ValueError("No access_token in token response")
       logger.info("Successfully obtained Dataminr access token.")
       return access_token
   
   def _load_state(bucket_name: str, state_key: str) -> dict:
       """Load the last cursor (alertId) from GCS."""
       try:
           bucket = storage_client.bucket(bucket_name)
           blob = bucket.blob(state_key)
           if blob.exists():
               data = json.loads(blob.download_as_text())
               logger.info(f"Loaded state: {data}")
               return data
       except Exception as e:
           logger.warning(f"State read error: {e}")
       logger.info("No previous state found.")
       return {}
   
   def _save_state(bucket_name: str, state_key: str, state: dict) -> None:
       """Save the cursor state to GCS."""
       bucket = storage_client.bucket(bucket_name)
       blob = bucket.blob(state_key)
       blob.upload_from_string(
           json.dumps(state), content_type="application/json"
       )
       logger.info(f"Saved state: {state}")
   
   def _fetch_alerts(
       access_token: str,
       alert_lists: str,
       page_size: int,
       cursor: str = None,
   ) -> tuple:
       """Fetch a page of alerts from the Dataminr Pulse API."""
       headers = {
           "Authorization": f"Bearer {access_token}",
           "Accept": "application/json",
       }
       params = {
           "lists": alert_lists,
           "num": page_size,
       }
       if cursor:
           params["from"] = cursor
   
       resp = requests.get(
           ALERTS_URL, headers=headers, params=params, timeout=60
       )
   
       # Handle rate limiting via response headers
       rate_remaining = resp.headers.get("x-ratelimit-remaining")
       rate_reset = resp.headers.get("x-ratelimit-reset")
   
       if resp.status_code == 429:
           reset_time = int(rate_reset) if rate_reset else 60
           wait_seconds = max(reset_time - int(time.time()), 1)
           logger.warning(
               f"Rate limited. Waiting {wait_seconds}s before retry."
           )
           time.sleep(wait_seconds)
           resp = requests.get(
               ALERTS_URL, headers=headers, params=params, timeout=60
           )
   
       resp.raise_for_status()
   
       if rate_remaining is not None:
           logger.info(
               f"Rate limit remaining: {rate_remaining}, reset: {rate_reset}"
           )
   
       data = resp.json()
       alerts = data if isinstance(data, list) else data.get("data", [])
       return alerts
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """Cloud Run function entry point triggered by Pub/Sub."""
       bucket_name = os.environ["GCS_BUCKET"]
       prefix = os.environ.get("GCS_PREFIX", "dataminr_alerts")
       state_key = os.environ.get("STATE_KEY", "dataminr_state/cursor.json")
       client_id = os.environ["CLIENT_ID"]
       client_secret = os.environ["CLIENT_SECRET"]
       alert_lists = os.environ["ALERT_LISTS"]
       max_records = int(os.environ.get("MAX_RECORDS", "1000"))
       page_size = min(int(os.environ.get("PAGE_SIZE", "40")), 40)
       lookback_hours = int(os.environ.get("LOOKBACK_HOURS", "24"))
   
       try:
           access_token = _get_access_token(client_id, client_secret)
           state = _load_state(bucket_name, state_key)
           cursor = state.get("last_cursor")
           is_first_run = cursor is None
   
           all_alerts = []
           total_fetched = 0
           pages_fetched = 0
   
           while total_fetched < max_records:
               logger.info(
                   f"Fetching page {pages_fetched + 1} (cursor: {cursor})..."
               )
               alerts = _fetch_alerts(
                   access_token, alert_lists, page_size, cursor=cursor
               )
   
               if not alerts:
                   logger.info("No more alerts returned. Stopping pagination.")
                   break
   
               # Filter by lookback window on first run (no prior cursor)
               if is_first_run:
                   cutoff_ms = int(
                       (
                           datetime.now(timezone.utc)
                           - timedelta(hours=lookback_hours)
                       ).timestamp()
                       * 1000
                   )
                   alerts = [
                       a for a in alerts if a.get("eventTime", 0) >= cutoff_ms
                   ]
   
               all_alerts.extend(alerts)
               total_fetched += len(alerts)
               pages_fetched += 1
   
               # Update cursor to the last alertId in this page
               last_alert = alerts[-1] if alerts else None
               if last_alert and "alertId" in last_alert:
                   cursor = last_alert["alertId"]
               else:
                   break
   
               # Stop if we received fewer alerts than requested
               if len(alerts) < page_size:
                   logger.info("Received partial page. Stopping pagination.")
                   break
   
           logger.info(
               f"Collected {len(all_alerts)} alerts across {pages_fetched} pages."
           )
   
           if not all_alerts:
               logger.info("No new alerts to write.")
               return "No new alerts", 200
   
           # Write alerts as NDJSON to GCS
           now_str = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
           blob_path = f"{prefix}/{now_str}.ndjson"
           ndjson_body = "\n".join(
               json.dumps(alert, separators=(",", ":")) for alert in all_alerts
           )
   
           bucket = storage_client.bucket(bucket_name)
           blob = bucket.blob(blob_path)
           blob.upload_from_string(
               ndjson_body, content_type="application/x-ndjson"
           )
   
           _save_state(
               bucket_name,
               state_key,
               {
                   "last_cursor": cursor,
                   "last_run": datetime.now(timezone.utc).isoformat(),
               },
           )
   
           msg = (
               f"Wrote {len(all_alerts)} alerts to "
               f"gs://{bucket_name}/{blob_path}"
           )
           logger.info(msg)
           return msg, 200
   
       except Exception as e:
           logger.error(f"Error collecting Dataminr alerts: {e}")
           raise
   

9. 点击部署。

10. 等待函数部署完毕。部署完成后，状态会变为绿色对勾标记。

配置环境变量

1. 部署函数后，前往 Cloud Run Functions > dataminr-alert-collector。
2. 点击修改和部署新的修订版本。
3. 点击变量和密钥标签页（或展开运行时、构建、连接和安全设置 [第 1 代]）。

4. 添加以下环境变量：

   键	示例值
   GCS_BUCKET	dataminr-alert-logs
   GCS_PREFIX	dataminr_alerts
   STATE_KEY	dataminr_state/cursor.json
   CLIENT_ID	您的 Dataminr OAuth 2.0 客户端 ID
   CLIENT_SECRET	您的 Dataminr OAuth 2.0 客户端密钥
   ALERT_LISTS	以英文逗号分隔的 Dataminr 提醒列表 ID
   MAX_RECORDS	1000
   PAGE_SIZE	40
   LOOKBACK_HOURS	24

   注意：对于生产部署，请将 CLIENT_ID 和 CLIENT_SECRET 存储在 Google Cloud Secret Manager 中，并使用引用 Secret 选项来引用它们，而不是使用纯文本环境变量。
   注意：PAGE_SIZE 值不得超过 40，这是 Dataminr Pulse API 支持的最大网页大小。ALERT_LISTS 值应包含您要监控的提醒列表 ID，并以英文逗号分隔。

5. 点击部署。

创建 Cloud Scheduler 作业

Cloud Scheduler 会按计划将消息发布到 Pub/Sub 主题，从而触发 Cloud Run 函数轮询 Dataminr Pulse 以获取新提醒。

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

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

   设置	值
   名称	dataminr-alert-poll
   区域	选择与函数相同的区域
   频率	*/5 * * * *（每 5 分钟）
   时区	选择您所在的时区（例如 UTC）

4. 点击继续。

5. 在配置执行部分中：

   • 目标类型：选择 Pub/Sub
   • 主题：选择 dataminr-alert-trigger
   • 消息正文：输入 {"poll": true}

6. 点击创建。

   注意： Dataminr Pulse 提供实时提醒。轮询频率为 5 分钟，可确保及时提取数据，同时遵守 API 速率限制。您可以根据自己的需求调整 Cron 表达式，以提高或降低轮询频率。

验证 Cloud Run 函数

1. 在 Cloud Scheduler 中，找到 dataminr-alert-poll 作业。
2. 点击 Force Run 以触发立即执行。
3. 依次前往 Cloud Run Functions > dataminr-alert-collector > 日志。

4. 通过检查日志条目（例如以下条目）来验证函数是否已成功执行：

   Successfully obtained Dataminr access token.
   Fetching page 1 (cursor: None)...
   Collected 35 alerts across 1 pages.
   Wrote 35 alerts to gs://dataminr-alert-logs/dataminr_alerts/20250115T103000Z.ndjson
   

5. 依次前往 Cloud Storage >“存储分区”>“dataminr-alert-logs”。

6. 前往 dataminr_alerts/ 前缀。

7. 验证是否正在使用 Dataminr 提醒数据创建 NDJSON 文件。

检索 Google SecOps 服务账号并配置 Feed

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

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

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

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

   [e-mail]
   

9. 复制此电子邮件地址以供下一部分使用。

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

10. 点击下一步。

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

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

      gs://dataminr-alert-logs/dataminr_alerts/
      

      注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

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

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

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

        注意： 如果您选择了删除选项，则服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。

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

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

    • 注入标签：要应用于此 Feed 中事件的标签（例如 DATAMINR_ALERT）。

12. 点击下一步。

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

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

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

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

6. 点击保存。

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

UDM 映射表

日志字段	UDM 映射	逻辑
alertId	metadata.product_log_id	直接复制值
alertType.color	about.labels.alertType_color	直接复制值
alertType.id	about.labels.alertType_id	直接复制值
alertType.name	about.labels.alertType_name	直接复制值
availableRelatedAlerts	about.labels.availableRelatedAlerts	已转换为字符串
caption	metadata.description	直接复制值
cat.name	security_result.category_details	直接复制值
cat.id	security_result.detection_fields.categories_id	直接复制值
cat.idStr	security_result.detection_fields.categories_idStr	直接复制值
cat.path	security_result.detection_fields.categories_path	直接复制值
cat.requested	security_result.detection_fields.categories_requested	直接复制值
cat.retired	security_result.detection_fields.categories_retired	已转换为字符串
cat.topicType	about.labels.categories_topicType	直接复制值
cat.name	security_result.category	如果 cat.name == "Cybersecurity - Policy"，则设置为 POLICY_VIOLATION；如果属于 ["Cybersecurity - Threats & Vulnerabilities", "Cybersecurity - Crime & Malicious Activity", "Threats & Precautions", "Threats"]，则设置为 NETWORK_MALICIOUS；如果 =~ "Cybersecurity"，则设置为 NETWORK_SUSPICIOUS；如果 =~ "Email and Web Servers"，则设置为 MAIL_PHISHING；如果 =~ "Data Exposure and Breaches"，则设置为 DATA_EXFILTRATION；如果 =~ "Government, Policy, & Political Affairs"，则设置为 POLICY_VIOLATION；如果 =~ "(Malware
comp.dm_bucket.name	security_result.about.resource.attribute.labels.dmbucket%{bucket.id}	直接复制值
comp.dm_sector.name	security_result.about.resource.attribute.labels.dmsector%{sector.id}	直接复制值
comp.id	security_result.about.resource.attribute.labels.companies_id	直接复制值
comp.idStr	security_result.about.resource.attribute.labels.companies_idStr	直接复制值
comp.locations.city	security_result.about.location.city	如果 loc_index == 0，则值为 loc.city
comp.locations.country, comp.locations.state.symbol	security_result.about.location.country_or_region	如果 loc_index == 0 且两者均不为空，则串联为 %{loc.country} - %{loc.state.symbol}
comp.locations.postalCode	security_result.about.resource.attribute.labels.locations_postalCode	如果 loc_index 为 0 且不为空，则直接复制值
comp.locations.state.name	security_result.about.location.state	如果 loc_index == 0，则直接复制值
comp.locations.city	about.labels.loc_%{loc_index}_city	如果 loc_index != 0 且不为空，则直接复制值
comp.locations.country, comp.locations.state.symbol	about.labels.loc_%{loc_index}_country_or_region	如果 loc_index != 0 且两者均不为空，则串联为 %{loc.country} - %{loc.state.symbol}
comp.locations.postalCode	securityresult.about.resource.attribute.labels.locations%{loc_index}_postalCode	如果 loc_index != 0 且不为空，则直接复制值
comp.locations.state.name	about.labels.loc_%{loc_index}_state_name	如果 loc_index != 0 且不为空，则直接复制值
comp.name	security_result.about.resource.name	直接复制值
comp.requested	security_result.about.resource.attribute.labels.companies_requested	直接复制值
comp.retired	security_result.about.resource.attribute.labels.companies_retired	已转换为字符串
comp.ticker	security_result.about.resource.attribute.labels.companies_ticker	直接复制值
comp.topicType	security_result.about.resource.attribute.labels.companies_topicType	直接复制值
eventLocation.coordinates.0	principal.location.region_coordinates.latitude	直接复制值
eventLocation.coordinates.1	principal.location.region_coordinates.longitude	直接复制值
eventLocation.name	principal.location.name	直接复制值
eventLocation.places	principal.labels.location_places	使用英文逗号分隔符从数组联接
eventLocation.probability	principal.labels.eventLocation_probability	已转换为字符串
eventLocation.radius	principal.labels.eventLocation_radius	已转换为字符串
eventMapLargeURL	principal.labels.eventMapLargeURL	直接复制值
eventMapSmallURL	principal.labels.eventMapSmallURL	直接复制值
eventTime	@timestamp	从纪元毫秒转换为时间戳
eventVolume	about.labels.eventVolume	已转换为字符串
expandAlertURL	metadata.url_back_to_product	直接复制值
expandMapURL	principal.labels.expandMapURL	直接复制值
headerColor	about.labels.headerColor	直接复制值
headerLabel	about.labels.headerLabel	直接复制值
metadata.cyber.addresses.ip	principal.ip	如果 index == 0，则使用 grok 模式提取
metadata.cyber.addresses.port	principal.port	如果 index == 0，则直接复制值；否则，转换为整数
metadata.cyber.addresses.port	principal.labels.addresses_%{index}_port	如果索引不等于 0，则直接复制值
metadata.cyber.addresses.version	principal.labels.metadata_cyberaddresses%{index}_version	直接复制值
metadata.cyber.asns	network.asn	如果 index == 0，则直接复制值
metadata.cyber.asns	about.labels.metadatacyber%{index}_asn	如果索引不等于 0，则直接复制值
metadata.cyber.hashValues.value	security_result.about.file.sha1	如果 type == SHA1，则直接复制值，并转换为小写
metadata.cyber.hashValues.value	security_result.about.file.sha256	如果 type == SHA256，则直接复制值，并转换为小写
metadata.cyber.malwares	security_result.associations.name	直接复制值
metadata.cyber.malwares	security_result.associations.type	设置为 MALWARE
metadata.cyber.orgs	network.organization_name	如果 index == 0，则直接复制值
metadata.cyber.orgs	about.labels.metadatacyber%{index}_orgs	如果索引不等于 0，则直接复制值
metadata.cyber.products	principal.application	如果 index == 0，则直接复制值
metadata.cyber.products	principal.labels.metadata_cyberproducts%{index}	如果索引不等于 0，则直接复制值
metadata.cyber.threats	security_result.threat_name	如果 index == 0，则直接复制值
metadata.cyber.threats	security_result.about.labels.metadata_cyberthreats%{index}	如果索引不等于 0，则直接复制值
metadata.cyber.URLs	security_result.about.url	如果 index == 0，则直接复制值
metadata.cyber.URLs	securityresult.about.labels.url%{index}	如果索引不等于 0，则直接复制值
metadata.cyber.malwares.0	security_result.category	如果存在，则设置为 SOFTWARE_MALICIOUS
metadata.cyber.vulnerabilities.cvss	extensions.vulns.vulnerabilities.cvss_base_score	直接复制值
metadata.cyber.vulnerabilities.exploitPocLinks	extensions.vulns.vulnerabilities.cve_description	使用“n”分隔符从数组联接
metadata.cyber.vulnerabilities.id	extensions.vulns.vulnerabilities.cve_id	直接复制值
metadata.cyber.vulnerabilities.products.productName	extensions.vulns.vulnerabilities.about.application	如果 index == 0，则直接复制值
metadata.cyber.vulnerabilities.products.productVendor	extensions.vulns.vulnerabilities.vendor	如果 index == 0，则直接复制值
metadata.cyber.vulnerabilities.products.productVersion	extensions.vulns.vulnerabilities.about.platform_version	如果 index == 0，则直接复制值，并移除空格
metadata.cyber.vulnerabilities.products.productName	extensions.vulns.vulnerabilities.about.labels.productName_%{index}	如果索引不等于 0，则直接复制值
metadata.cyber.vulnerabilities.products.productVendor	extensions.vulns.vulnerabilities.about.labels.productVendor_%{index}	如果索引不等于 0，则直接复制值
metadata.cyber.vulnerabilities.products.productVersion	extensions.vulns.vulnerabilities.about.labels.productVersion_%{index}	如果 index != 0，则直接复制值，并移除空格
parentAlertId	about.labels.parentAlertId	直接复制值
post.languages.lang	target.labels.post_languageslang%{index}	直接复制值
post.languages.position	target.labels.post_languagesposition%{index}	已转换为字符串
post.link	target.labels.post_link	直接复制值
post.media.link	principal.resource.name	如果 index == 0，则直接复制值
post.media.description	target.resource.attribute.labels.post_media_description	如果 index == 0，则直接复制值
post.media.display_url	target.resource.attribute.labels.post_media_display_url	如果 index == 0，则直接复制值
post.media.isSafe	target.resource.attribute.labels.post_media_isSafe	如果 index == 0，则转换为字符串
post.media.media_url	target.resource.attribute.labels.post_media_media_url	如果 index == 0，则直接复制值
post.media.sizes.large.h	target.resource.attribute.labels.post_media_sizes_large_h	如果 index == 0，则转换为字符串
post.media.sizes.large.resize	target.resource.attribute.labels.post_media_sizes_large_resize	如果 index == 0，则直接复制值
post.media.sizes.large.w	target.resource.attribute.labels.post_media_sizes_large_w	如果 index == 0，则转换为字符串
post.media.sizes.medium.h	target.resource.attribute.labels.post_media_sizes_medium_h	如果 index == 0，则转换为字符串
post.media.sizes.medium.resize	target.resource.attribute.labels.post_media_sizes_medium_resize	如果 index == 0，则直接复制值
post.media.sizes.medium.w	target.resource.attribute.labels.post_media_sizes_medium_w	如果 index == 0，则转换为字符串
post.media.sizes.small.h	target.resource.attribute.labels.post_media_sizes_small_h	如果 index == 0，则转换为字符串
post.media.sizes.small.resize	target.resource.attribute.labels.post_media_sizes_small_resize	如果 index == 0，则直接复制值
post.media.sizes.small.w	target.resource.attribute.labels.post_media_sizes_small_w	如果 index == 0，则转换为字符串
post.media.sizes.thumb.h	target.resource.attribute.labels.post_media_sizes_thumb_h	如果 index == 0，则转换为字符串
post.media.sizes.thumb.resize	target.resource.attribute.labels.post_media_sizes_thumb_resize	如果 index == 0，则直接复制值
post.media.sizes.thumb.w	target.resource.attribute.labels.post_media_sizes_thumb_w	如果 index == 0，则转换为字符串
post.media.source	target.resource.attribute.labels.post_media_source	如果 index == 0，则直接复制值
post.media.thumbnail	target.resource.attribute.labels.post_media_thumbnail	如果 index == 0，则直接复制值
post.media.title	target.resource.attribute.labels.post_media_title	如果 index == 0，则直接复制值
post.media.url	target.resource.attribute.labels.post_media_url	如果 index == 0，则直接复制值
post.media.video_info.duration_millis	target.resource.attribute.labels.post_media_video_info_duration_millis	如果 index == 0，则转换为字符串
post.media.video_info.aspect_ratio	target.resource.attribute.labels.post_media_video_info_aspect_ratio	如果 index == 0，则串联为 %{med.video_info.aspect_ratio.0}、%{med.video_info.aspect_ratio.1}
post.media.video_info.variants.bitrate	target.resource.attribute.labels.post_media_video_info_variantsbitrate%{var_index}	已转换为字符串
post.media.video_info.variants.content_type	target.resource.attribute.labels.post_media_video_info_variants_contenttype%{var_index}	直接复制值
post.media.video_info.variants.url	target.resource.attribute.labels.post_media_video_info_variantsurl%{var_index}	直接复制值
post.media.type	principal.resource.resource_subtype	如果 index == 0，则直接复制值
post.media.link	about.resource.name	如果索引不等于 0，则直接复制值
post.media.description	about.resource.attribute.labels.post_media_description	如果索引不等于 0，则直接复制值
post.media.display_url	about.resource.attribute.labels.post_media_display_url	如果索引不等于 0，则直接复制值
post.media.isSafe	about.resource.attribute.labels.post_media_isSafe	如果 index != 0，则转换为字符串
post.media.media_url	about.resource.attribute.labels.post_media_media_url	如果索引不等于 0，则直接复制值
post.media.sizes.large.h	about.resource.attribute.labels.post_media_sizes_large_h	如果 index != 0，则转换为字符串
post.media.sizes.large.resize	about.resource.attribute.labels.post_media_sizes_large_resize	如果索引不等于 0，则直接复制值
post.media.sizes.large.w	about.resource.attribute.labels.post_media_sizes_large_w	如果 index != 0，则转换为字符串
post.media.sizes.medium.h	about.resource.attribute.labels.post_media_sizes_medium_h	如果 index != 0，则转换为字符串
post.media.sizes.medium.resize	about.resource.attribute.labels.post_media_sizes_medium_resize	如果索引不等于 0，则直接复制值
post.media.sizes.medium.w	about.resource.attribute.labels.post_media_sizes_medium_w	如果 index != 0，则转换为字符串
post.media.sizes.small.h	about.resource.attribute.labels.post_media_sizes_small_h	如果 index != 0，则转换为字符串
post.media.sizes.small.resize	about.resource.attribute.labels.post_media_sizes_small_resize	如果索引不等于 0，则直接复制值
post.media.sizes.small.w	about.resource.attribute.labels.post_media_sizes_small_w	如果 index != 0，则转换为字符串
post.media.sizes.thumb.h	about.resource.attribute.labels.post_media_sizes_thumb_h	如果 index != 0，则转换为字符串
post.media.sizes.thumb.resize	about.resource.attribute.labels.post_media_sizes_thumb_resize	如果索引不等于 0，则直接复制值
post.media.sizes.thumb.w	about.resource.attribute.labels.post_media_sizes_thumb_w	如果 index != 0，则转换为字符串
post.media.source	about.resource.attribute.labels.post_media_source	如果索引不等于 0，则直接复制值
post.media.thumbnail	about.resource.attribute.labels.post_media_thumbnail	如果索引不等于 0，则直接复制值
post.media.title	about.resource.attribute.labels.post_media_title	如果索引不等于 0，则直接复制值
post.media.url	about.resource.attribute.labels.post_media_url	如果索引不等于 0，则直接复制值
post.media.video_info.duration_millis	about.resource.attribute.labels.post_media_video_info_duration_millis	如果 index != 0，则转换为字符串
post.media.video_info.aspect_ratio	about.resource.attribute.labels.post_media_video_info_aspect_ratio	如果 index != 0，则串联为 %{med.video_info.aspect_ratio.0}、%{med.video_info.aspect_ratio.1}
post.media.video_info.variants.bitrate	about.resource.attribute.labels.post_media_video_info_variantsbitrate%{var_index}	已转换为字符串
post.media.video_info.variants.content_type	about.resource.attribute.labels.post_media_video_info_variants_contenttype%{var_index}	直接复制值
post.media.video_info.variants.url	about.resource.attribute.labels.post_media_video_info_variantsurl%{var_index}	直接复制值
post.media.type	about.resource.resource_subtype	如果索引不等于 0，则直接复制值
post.translatedText	target.labels.post_translatedText	直接复制值
post.text	target.labels.post_text	直接复制值
post.timestamp	target.resource.attribute.creation_time	从纪元毫秒转换为时间戳
publisherCategory.color	target.labels.publisherCategory_color	直接复制值
publisherCategory.name	target.labels.publisherCategory_name	直接复制值
publisherCategory.shortName	target.labels.publisherCategory_shortName	直接复制值
relatedTerms.url	principal.labels.relatedTerms_%{terms.text}	直接复制值
relatedTermsQueryURL	principal.labels.relatedTermsQueryURL	直接复制值
sect.id	about.labels.sectors_id	直接复制值
sect.idStr	about.labels.sectors_idStr	直接复制值
sect.name	about.labels.sectors_name	直接复制值
sect.retired	about.labels.sectors_retired	已转换为字符串
sect.topicType	about.labels.sectors_topicType	直接复制值
source.channels.0	principal.application	直接复制值
source.displayName	principal.user.user_display_name	直接复制值
source.link	principal.url	直接复制值
source.verified	principal.labels.source_verified	已转换为字符串
subCaption.bullets.content	about.labels.subCaption_bullets_content	直接复制值
subCaption.bullets.media	about.labels.subCaption_bullets_media	直接复制值
subCaption.bullets.source	about.labels.subCaption_bullets_source	直接复制值
watchlist.id	about.labels.watchlistsMatchedByType_id	直接复制值
watchlist.externalTopicIds	about.labels.watchlistsMatchedByType_externalTopicIds	使用英文逗号分隔符从数组联接
watchlist.name	about.labels.watchlistsMatchedByType_name	直接复制值
watchlist.type	about.labels.watchlistsMatchedByType_type	直接复制值
watchlist.userProperties.omnilist	about.labels.watchlistsMatchedByType_userProperties_omnilist	直接复制值
watchlist.userProperties.uiListType	about.labels.watchlistsMatchedByType_userProperties_uiListType	直接复制值
watchlist.userProperties.watchlistColor	about.labels.watchlistsMatchedByType_userProperties_watchlistColor	直接复制值
watchlist.locationGroups.locations.id	about.labels.watchlistsMatchedByTypelocationGroups%{lg_i}_locationsid%{loc_i}	直接复制值
watchlist.locationGroups.locations.lng	about.labels.watchlistsMatchedByTypelocationGroups%{lg_i}_locationslng%{loc_i}	如果 lg_i != 0 或 loc_i != 0，则转换为字符串
watchlist.locationGroups.locations.lat	about.labels.watchlistsMatchedByTypelocationGroups%{lg_i}_locationslat%{loc_i}	如果 lg_i != 0 或 loc_i != 0，则转换为字符串
watchlist.locationGroups.locations.name	about.labels.watchlistsMatchedByTypelocationGroups%{lg_i}_locationsname%{loc_i}	如果 lg_i != 0 或 loc_i != 0，则直接复制值
watchlist.locationGroups.id	about.labels.watchlistsMatchedByType_locationGroupsid%{lg_i}	直接复制值
watchlist.locationGroups.name	about.labels.watchlistsMatchedByType_locationGroupsname%{lg_i}	直接复制值
watchlist.locationGroups.locations.lng	about.location.region_coordinates.longitude	如果 lg_i == 0 且 loc_i == 0，则直接复制值
watchlist.locationGroups.locations.lat	about.location.region_coordinates.latitude	如果 lg_i == 0 且 loc_i == 0，则直接复制值
watchlist.locationGroups.locations.name	about.location.name	如果 lg_i == 0 且 loc_i == 0，则直接复制值
source.entityName	principal.hostname	直接复制值
	metadata.event_type	设置为“GENERIC_EVENT”；如果 principal_ip 或 principal.hostname 不为空，则更改为“SCAN_HOST”

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