收集 Tenable 审核日志

支持的平台：

本文档介绍了如何使用 Cloud Storage V2 将 Tenable 审核日志注入到 Google Security Operations。

Tenable Vulnerability Management（以前称为 Tenable.io）是一个基于云的漏洞管理平台 (cloud.tenable.com)，其活动日志会记录用户身份验证、API 访问、配置更改和管理操作。Tenable Vulnerability Management REST API 提供对这些活动日志事件的程序化访问。

准备工作

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

  • Google SecOps 实例
  • 启用了 Cloud Storage API 的 Google Cloud 项目
  • 创建和管理 Cloud Storage 存储分区的权限
  • 管理 Cloud Storage 存储分区的 Identity and Access Management (IAM) 政策的权限
  • 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
  • 拥有对 Tenable Vulnerability Management (cloud.tenable.com) 的特权访问权限，并具有管理员角色
  • 在具有管理员角色的用户账号上生成的 Tenable Vulnerability Management API 密钥（访问密钥和私有密钥）

创建 Cloud Storage 存储桶

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

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

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

  6. 点击创建。

收集 Tenable Vulnerability Management API 凭据

生成 API 密钥

  1. 登录 Tenable Vulnerability Management。
  2. 在任意页面的右上角，点击蓝色用户圆圈，然后点击我的个人资料。系统会显示我的账号页面。
  3. 前往 API 密钥标签页。

  4. 点击生成。系统随即会显示生成 API 密钥窗口，其中包含一条警告。

    “注意：生成密钥会替换相应用户账号中的所有现有 API 密钥，包括其他集成已使用的密钥。请为此集成使用专用用户账号，或更新使用之前密钥的每个应用。”

  5. 查看警告，然后点击生成。

  6. 复制以下详细信息并将其保存在安全的位置：

    • 访问密钥：API 访问密钥
    • Secret Key：API Secret 密钥

验证 API 访问权限

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

    # Replace with your actual credentials
ACCESS_KEY="your-access-key"
SECRET_KEY="your-secret-key"

curl -s -H "X-ApiKeys: accessKey=${ACCESS_KEY};secretKey=${SECRET_KEY}" \
    "https://cloud.tenable.com/audit-log/v1/events?limit=1" | head -c 500

验证权限

如需验证账号是否具有所需权限，请执行以下操作：

  1. 登录 Tenable Vulnerability Management。
  2. 在左侧导航栏中，点击设置。
  3. 点击访问权限控制板块。
  4. 点击用户标签页，然后点击用于此集成的用户账号。

  5. 验证相应账号是否拥有管理员角色。活动日志端点需要“管理员”用户角色：任何自定义角色都无法授予对活动日志的访问权限，而任何其他角色都会收到 HTTP 403 响应。

  6. 如果您没有所需的权限，请与您的 Tenable Vulnerability Management 管理员联系。

为 Cloud Run 函数创建服务账号

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

创建服务账号

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

必须拥有这些角色，才能：

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

授予对 Cloud Storage 存储桶的 IAM 权限

向服务账号授予对 Cloud Storage 存储桶的写入权限：

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

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Tenable Vulnerability Management REST API 中提取日志并将其写入 Cloud Storage。

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

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

    设置 值
    Service 名称 tenable-audit-collector
    区域 选择与您的 Cloud Storage 存储桶匹配的区域（例如 us-central1）
    运行时 选择 Python 3.12 或更高版本

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

    1. 点击 + 添加触发器。
    2. 选择 Cloud Pub/Sub。
    3. 在选择 Cloud Pub/Sub 主题部分，选择主题 tenable-audit-trigger。
    4. 点击保存。

  6. 在身份验证部分中：

    1. 选择需要进行身份验证。
    2. 检查 Identity and Access Management (IAM)。

  7. 向下滚动并展开容器、网络、安全性。

  8. 前往安全性标签页：

    • 服务账号：选择服务账号 tenable-audit-collector-sa。

  9. 前往容器标签页：

    1. 点击变量和密钥。
    2. 为每个环境变量点击+ 添加变量：
    变量名称 示例值 说明
    GCS_BUCKET tenable-audit-logs Cloud Storage 存储桶名称
    GCS_PREFIX tenable 日志文件的前缀
    STATE_KEY tenable-state.json 状态路径，位于日志前缀之外
    TENABLE_ACCESS_KEY your-access-key Tenable Vulnerability Management 访问密钥
    TENABLE_SECRET_KEY your-secret-key Tenable Vulnerability Management Secret 密钥
    MAX_RECORDS 5000 每次运行的记录数上限
    PAGE_SIZE 1000 每页记录数
    LOOKBACK_HOURS 24 初始回溯期
    OVERLAP_MINUTES 2 在水印之前重新读取的分钟数，用于捕获延迟编入索引的事件

  10. 在变量和密钥部分中，前往请求：

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

  11. 前往设置标签页：

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

  12. 在修订版本伸缩部分中：

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

  13. 点击创建。

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

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

添加函数代码

  1. 在入口点字段中输入 main。

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

    • 第一个文件 - main.py：
    import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import json
import os
import re
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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'tenable')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'tenable-state.json')
TENABLE_ACCESS_KEY = os.environ.get('TENABLE_ACCESS_KEY')
TENABLE_SECRET_KEY = os.environ.get('TENABLE_SECRET_KEY')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
# The audit-log API accepts a limit up to 10000.
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events
# Tenable indexes late are still collected. Re-read events are dropped by id.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

TENABLE_API_BASE = 'https://cloud.tenable.com'

class FetchError(Exception):
  """Raised when the Tenable API call fails.

  The watermark must never advance on a failed fetch, otherwise every event in
  the failed window is skipped permanently.
  """

def parse_datetime(value: str) -> datetime:
  """Parse a Tenable `received` timestamp into an aware datetime.

  Tenable returns ISO 8601 with either whole seconds (2018-12-31T23:09:40Z) or
  fractional seconds (2024-01-16T15:12:47.334Z). Fractional digits are trimmed
  to microseconds because fromisoformat accepts at most six.
  """
  text = str(value).strip()
  if text.endswith('Z'):
    text = text[:-1] + '+00:00'
  text = re.sub(r'\.(\d{6})\d+', r'.\1', text)
  dt = datetime.fromisoformat(text)
  if dt.tzinfo is None:
    dt = dt.replace(tzinfo=timezone.utc)
  return dt.astimezone(timezone.utc)

@functions_framework.cloud_event
def main(cloud_event):
  """Fetch Tenable Vulnerability Management activity log events and write them to Cloud Storage.

  Args:
    cloud_event: CloudEvent object containing the Pub/Sub message.
  """
  if not all([GCS_BUCKET, TENABLE_ACCESS_KEY, TENABLE_SECRET_KEY]):
    # Raise rather than return: a bare return acks the Pub/Sub message and
    # reports the run as successful, silently discarding the schedule tick.
    raise RuntimeError('Missing required environment variables')

  bucket = storage_client.bucket(GCS_BUCKET)
  state = load_state(bucket, STATE_KEY)

  now = datetime.now(timezone.utc)
  watermark = None
  if state.get('last_event_time'):
    watermark = parse_datetime(state['last_event_time'])
  seen_ids = set(state.get('seen_ids', []))

  if watermark is None:
    start_time = now - timedelta(hours=LOOKBACK_HOURS)
  else:
    start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

  print(f"Fetching logs from {start_time.isoformat()} to {now.isoformat()}")

  # A FetchError propagates: the run fails, the watermark is untouched, and the
  # next invocation retries the same window.
  records, newest_event_time = fetch_logs(
    start_time=start_time,
    page_size=PAGE_SIZE,
    max_records=MAX_RECORDS,
  )

  # Drop events already written by an earlier run. Without this the overlap
  # window re-emits its events on every invocation, and an idle tenant has its
  # newest events rewritten to a new object every hour.
  fresh = [r for r in records if str(r.get('id', '')) not in seen_ids]
  print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

  if not fresh:
    print("No new log records found. Watermark left unchanged.")
    return

  if not newest_event_time:
    raise FetchError('Records were returned but no `received` timestamp could be parsed')

  timestamp = now.strftime('%Y%m%dT%H%M%SZ')
  object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
  blob = bucket.blob(object_key)

  ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
  blob.upload_from_string(ndjson, content_type='application/x-ndjson')

  print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")

  # Advance the watermark only after the data is durably written.
  new_watermark = parse_datetime(newest_event_time)
  if watermark and new_watermark < watermark:
    new_watermark = watermark

  # Retain only the ids still inside the overlap window, so the state file
  # stays small while covering every event the next query can re-read.
  cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
  retained = []
  for record in records:
    received = record.get('received')
    if not received:
      continue
    try:
      if parse_datetime(received) >= cutoff:
        retained.append(str(record.get('id', '')))
    except ValueError:
      continue

  save_state(bucket, STATE_KEY, {
    'last_event_time': new_watermark.isoformat(),
    'seen_ids': sorted(i for i in set(retained) if i),
  })

  print(f"Successfully processed {len(fresh)} records")

def load_state(bucket, key):
  """Read the collector state from Cloud Storage.

  Only a missing object is treated as a cold start. Any other error is raised:
  swallowing it would silently reset collection to the full lookback window and
  re-ingest that entire period.
  """
  blob = bucket.blob(key)
  try:
    return json.loads(blob.download_as_text())
  except NotFound:
    print('No state file found. Starting from the lookback window.')
    return {}

def save_state(bucket, key, state: dict):
  """Write the collector state to Cloud Storage.

  Failures are raised, not logged. If the state write fails after the data was
  uploaded, the next run repeats the same window and duplicates it.
  """
  blob = bucket.blob(key)
  blob.upload_from_string(
    json.dumps(state, indent=2),
    content_type='application/json',
  )
  print(f"Saved state: last_event_time={state.get('last_event_time')}")

def fetch_logs(start_time: datetime, page_size: int, max_records: int):
  """Fetch audit events from the Tenable Vulnerability Management API.

  Uses offset pagination, which is what the endpoint documents: the request
  accepts `limit` (maximum 10000) and `offset`, and the response `pagination`
  object returns `offset`, `limit`, `count` and `total`.

  Args:
    start_time: Exclusive lower bound for the `received` timestamp
    page_size: Records per page (the API accepts up to 10000)
    max_records: Maximum total records to fetch in one run

  Returns:
    Tuple of (records list, newest `received` value as an ISO 8601 string).

  Raises:
    FetchError: on any API or transport failure, so the caller cannot mistake
      a failed fetch for an empty result and advance the watermark.
  """
  endpoint = f"{TENABLE_API_BASE}/audit-log/v1/events"

  headers = {
    'X-ApiKeys': f'accessKey={TENABLE_ACCESS_KEY};secretKey={TENABLE_SECRET_KEY}',
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-TenableAuditCollector/1.0'
  }

  records = []
  newest_time = None
  page_num = 0
  backoff = 1.0
  rate_limit_retries = 0
  MAX_RATE_LIMIT_RETRIES = 5
  offset = 0

  while True:
    page_num += 1

    if len(records) >= max_records:
      # Stop cleanly. The remaining events stay ahead of the watermark and
      # are collected by the next run.
      print(f"Reached max_records limit ({max_records})")
      break

    # Every parameter is rebuilt per page. `sort=received:asc` is required:
    # a time watermark is only correct over an ascending scan, otherwise a
    # truncated run advances the watermark past events it never read.
    params = {
      'f': f'date.gt:{start_time.strftime("%Y-%m-%dT%H:%M:%SZ")}',
      'sort': 'received:asc',
      'limit': min(page_size, max_records - len(records)),
      'offset': offset,
    }
    url = f"{endpoint}?" + '&'.join(f"{k}={v}" for k, v in params.items())

    try:
      response = http.request('GET', url, headers=headers)
    except Exception as e:
      raise FetchError(f'Request to {endpoint} failed: {e}') from e

    if response.status == 429:
      rate_limit_retries += 1
      if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
        raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
      raw_retry_after = response.headers.get('Retry-After')
      try:
        # Retry-After may also be an HTTP date, which int() cannot parse.
        retry_after = int(raw_retry_after) if raw_retry_after else int(backoff)
      except (TypeError, ValueError):
        retry_after = int(backoff)
      print(f"Rate limited (429). Retrying after {retry_after}s...")
      time.sleep(retry_after)
      backoff = min(backoff * 2, 30.0)
      continue

    backoff = 1.0
    rate_limit_retries = 0

    if response.status != 200:
      body = response.data.decode('utf-8')
      raise FetchError(f'HTTP {response.status} from the Tenable audit log API: {body}')

    try:
      data = json.loads(response.data.decode('utf-8'))
    except json.JSONDecodeError as e:
      raise FetchError(f'Malformed JSON response from the Tenable audit log API: {e}') from e

    page_results = data.get('events', [])

    if not page_results:
      print("No more results (empty page)")
      break

    print(f"Page {page_num}: Retrieved {len(page_results)} events")
    records.extend(page_results)

    for event in page_results:
      received = event.get('received')
      if not received:
        continue
      try:
        if newest_time is None or parse_datetime(received) > parse_datetime(newest_time):
          newest_time = received
      except ValueError as e:
        print(f"Warning: Could not parse event time {received!r}: {e}")

    offset += len(page_results)
    total = data.get('pagination', {}).get('total')
    if total is not None and offset >= total:
      print("No more pages (all matching events retrieved)")
      break

  print(f"Retrieved {len(records)} total records from {page_num} pages")
  return records, newest_time

    • 第二个文件 - 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. 提供以下配置详细信息：

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

  4. 点击创建。

时间表频率选项

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

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

测试集成

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

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

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X events
Fetched X records, Y new after deduplication
Wrote Y records to gs://tenable-audit-logs/tenable/logs_YYYYMMDDTHHMMSSZ.ndjson
Saved state: last_event_time=YYYY-MM-DDTHH:MM:SS+00:00
Successfully processed Y records

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

  9. 点击您的存储桶名称 (tenable-audit-logs)。

  10. 转到 tenable/ 文件夹。

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

在没有新内容到达的运行中，该函数会记录 No new log records found. Watermark left unchanged.，但不写入任何对象。这是预期行为：重新写入相同的事件会在 Google SecOps 中复制这些事件。

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

  • HTTP 401：检查环境变量中的 API 密钥
  • HTTP 403：相应账号不是管理员账号。活动日志端点需要管理员 [64] 用户角色。
  • HTTP 429：速率限制。该函数会使用退避机制进行重试，然后在不推进水位的情况下使运行失败，因此不会跳过任何事件。
  • 缺少环境变量：检查是否已设置所有必需的变量

在 Google SecOps 中配置 Feed 以注入 Tenable 审核日志

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

  7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：

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

  8. 复制此电子邮件地址。

  9. 点击下一步。

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

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

      gs://tenable-audit-logs/tenable/
      • 替换：
        • tenable-audit-logs：您的 Cloud Storage 存储桶名称。
        • tenable：存储日志的可选前缀/文件夹路径（留空表示根目录）。

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

      • 永不删除文件：永不从源中删除文件（建议用于测试）。

      • 删除已转移的文件和空目录：在成功提取完成后，从来源中删除文件和空目录。

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

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

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

  11. 点击下一步。

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

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

Google SecOps 服务账号需要您的 Cloud Storage 存储桶具备两个角色：用于读取日志对象的 Storage Object Viewer，以及用于读取存储桶元数据的存储桶级角色。

  1. 前往 Cloud Storage > 存储分区。
  2. 点击您的存储桶名称。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
    • 分配角色：选择以下两个角色：
      • Storage Object Viewer：读取日志对象。
      • Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储空间旧版存储分区写入者，该角色也会授予删除权限。
  6. 点击保存。

UDM 映射表

日志字段 UDM 映射 逻辑
crud_label additional.fields 已合并
fields_label additional.fields 已合并
field_name extensions.auth.mechanism 已映射：X-Access-Type → mech_label
mech_label extensions.auth.mechanism 已合并
extension_value extensions.auth.type 直接映射
description metadata.description 直接映射
received metadata.event_timestamp 解析为 ISO8601
has_principal metadata.event_type 映射的值（总共 5 个，例如 true → USER_LOGIN、true → USER_CREATION、true → `USER…
has_user metadata.event_type 已映射：true → USER_UNCATEGORIZED
action metadata.product_event_type 直接映射
id metadata.product_log_id 直接映射
field_name principal.asset.ip 已映射：X-Forwarded-For → ip
ip principal.asset.ip 已合并
field_name principal.ip 已映射：X-Forwarded-For → ip
ip principal.ip 已合并
actor.name principal.user.email_addresses 已合并
actor.id principal.user.userid 直接映射
AUTH_VIOLOATION security_result.category 已合并
is_failure security_result.category 已映射：true → AUTH_VIOLOATION
is_anonymous_label security_result.detection_fields 已合并
is_failure_label security_result.detection_fields 已合并
target1.name target.user.email_addresses 已合并
target1.type target.user.role_name 直接映射
target1.id target.user.userid 直接映射
不适用 extensions.auth.type 常量：AUTHTYPE_UNSPECIFIED
不适用 metadata.event_type 常量：USER_LOGIN
不适用 metadata.product_name 常量：TENABLE AUDIT
不适用 metadata.vendor_name 常量：TENABLE

更新日志

查看相应解析器的更改日志

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