收集 Okta Advanced Server Access 日志

支持的平台：

本文档介绍了如何使用 Google Cloud Storage V2 将 Okta Advanced Server Access 日志注入到 Google Security Operations。

Okta Advanced Server Access（以前称为 ScaleFT）是一种特权访问管理解决方案，可提供对服务器的 SSH 和 RDP 访问权限，并具有临时证书和全面的审核日志记录功能。该平台会将审核事件保留 90 天，并提供一个 API 来检索这些事件，以便在 SIEM 平台中进行长期存储和分析。

准备工作

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

  • Google SecOps 实例
  • 启用了 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 GCP 项目
  • 创建和管理 GCS 存储桶的权限
  • 拥有管理 GCS 存储桶 IAM 政策的权限
  • 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
  • 对您的 Okta Advanced Server Access 团队拥有管理员访问权限
  • 能够在高级服务器访问中创建服务用户和 API 密钥

配置 Okta Advanced Server Access API 访问权限

如需使 Google SecOps 能够检索审核事件，您需要创建服务用户并生成 API 密钥。服务用户是专为自动化和 API 集成设计的特殊账号。

创建服务用户

  1. 登录高级服务器访问权限信息中心。
  2. 转到用户。
  3. 选择服务用户标签页。
  4. 点击创建服务用户。

  5. 在用户名字段中，输入服务用户的描述性名称（例如 chronicle-integration）。

  6. 点击 Create Service User（创建服务用户）。

生成 API 密钥

  1. 在服务用户详情页面上，点击创建 API 密钥。

  2. 立即复制并保存以下凭证：

    • API 密钥 ID：用于标识 API 密钥的 UUID（例如 6052868b-1b04-4a14-8288-e6496d7f2f4b）
    • API 密钥 Secret：用于对 API 请求进行身份验证的 Base64 编码 Secret 字符串

  3. 点击完成。

将服务用户分配到具有适当权限的群组

服务用户必须是具有适当团队级角色的群组的成员，才能访问 Audits API。

  1. 在 Advanced Server Access 信息中心内，前往群组。
  2. 选择现有群组或创建新群组以用于 API 访问。
  3. 点击群组名称以打开群组详情。
  4. 选择用户标签。
  5. 点击添加用户。
  6. 搜索并选择服务用户（例如 chronicle-integration）。
  7. 点击 Add（添加）。
  8. 选择角色标签页。

  9. 确保该群组拥有以下任一团队级角色：

    • reporting_user：对所有端点的只读权限（建议用于日志注入）
    • access_admin：完全管理员权限（授予所有 CRUD 权限）

  10. 如果未分配所需角色，请点击添加角色，选择 reporting_user 或 access_admin，然后点击保存。

验证权限

如需验证服务用户是否具有所需权限，请执行以下操作：

  1. 在 Advanced Server Access 信息中心内，前往群组。
  2. 点击包含服务用户的群组（例如 chronicle-integration）。
  3. 选择角色标签页。
  4. 确认列出了 reporting_user 或 access_admin 角色。
  5. 如果您看不到“角色”标签页，请与您的管理员联系以获取适当的访问权限。

测试 API 访问权限

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

    # Replace with your actual credentials
TEAM_NAME="your-team-name"
KEY_ID="your-api-key-id"
KEY_SECRET="your-api-key-secret"

# Obtain bearer token
TOKEN=$(curl -s -X POST "https://app.scaleft.com/v1/teams/${TEAM_NAME}/service_token" \
    -H "Content-Type: application/json" \
    -d "{\"key_id\": \"${KEY_ID}\", \"key_secret\": \"${KEY_SECRET}\"}" \
    | python3 -c "import sys,json; print(json.load(sys.stdin)['bearer_token'])")

# Test audit events access
curl -s -H "Authorization: Bearer ${TOKEN}" \
    "https://app.scaleft.com/v1/teams/${TEAM_NAME}/auditsV2?count=5" \
    | python3 -m json.tool

成功的响应会返回一个包含近期审核事件的 JSON 数组。

创建 Google Cloud Storage 存储桶

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

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

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

  6. 点击创建。

为 Cloud Run 函数创建服务账号

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

授予对 GCS 存储桶的 IAM 权限

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

创建 Pub/Sub 主题

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

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Okta Advanced Server Access Audits API 中提取日志并将其写入 GCS。

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

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

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

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

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

  6. 在身份验证部分中：

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

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

  8. 前往安全标签页：

    • 服务账号：选择 okta-asa-collector-sa

  9. 前往容器标签页：

    1. 点击变量和密钥。
    2. 为每个环境变量点击 + 添加变量：
    变量名称 示例值 说明
    GCS_BUCKET okta-asa-audit-logs GCS 存储桶名称
    GCS_PREFIX okta-asa-audit 日志文件的前缀
    STATE_KEY okta-asa-audit/state.json 状态文件路径
    ASA_TEAM_NAME your-team-name Advanced Server Access 团队名称
    ASA_KEY_ID your-api-key-id 服务用户 API 密钥 ID
    ASA_KEY_SECRET your-api-key-secret 服务用户 API 密钥
    MAX_RECORDS 5000 每次运行的最大记录数
    PAGE_SIZE 1000 每个 API 页面的记录数（最多 1000 条）
    LOOKBACK_HOURS 24 初始回溯期

  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
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time

http = urllib3.PoolManager(
  timeout=urllib3.Timeout(connect=10.0, read=60.0),
  retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'okta-asa-audit')
STATE_KEY = os.environ.get('STATE_KEY', 'okta-asa-audit/state.json')
ASA_TEAM_NAME = os.environ.get('ASA_TEAM_NAME')
ASA_KEY_ID = os.environ.get('ASA_KEY_ID')
ASA_KEY_SECRET = os.environ.get('ASA_KEY_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

ASA_API_BASE = 'https://app.scaleft.com'

@functions_framework.cloud_event
def main(cloud_event):
  if not all([GCS_BUCKET, ASA_TEAM_NAME, ASA_KEY_ID, ASA_KEY_SECRET]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket)
    now = datetime.now(timezone.utc)

    if isinstance(state, dict) and state.get('last_offset'):
      last_offset = state['last_offset']
    else:
      last_offset = None

    print(f"Starting audit log collection (last_offset={last_offset})")

    token = get_bearer_token()

    records, newest_offset = fetch_audit_events(
      token=token,
      last_offset=last_offset,
      page_size=PAGE_SIZE,
      max_records=MAX_RECORDS,
    )

    if not records:
      print("No new audit records found.")
      return

    timestamp = now.strftime('%Y%m%d_%H%M%S')
    object_key = f"{GCS_PREFIX}/okta_asa_audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

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

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

    if newest_offset:
      save_state(bucket, newest_offset)

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

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

def get_bearer_token():
  url = f"{ASA_API_BASE}/v1/teams/{ASA_TEAM_NAME}/service_token"
  body = json.dumps({
    'key_id': ASA_KEY_ID,
    'key_secret': ASA_KEY_SECRET
  }).encode('utf-8')

  response = http.request(
    'POST', url,
    body=body,
    headers={'Content-Type': 'application/json'}
  )

  if response.status != 200:
    raise Exception(
      f"Service token request failed: {response.status} - "
      f"{response.data.decode('utf-8')}"
    )

  data = json.loads(response.data.decode('utf-8'))
  token = data.get('bearer_token')
  if not token:
    raise Exception("No bearer_token in service token response")

  print("Successfully obtained ASA bearer token")
  return token

def fetch_audit_events(token, last_offset, page_size, max_records):
  endpoint = f"{ASA_API_BASE}/v1/teams/{ASA_TEAM_NAME}/auditsV2"

  headers = {
    'Authorization': f'Bearer {token}',
    'Accept': 'application/json',
  }

  records = []
  page_num = 0
  backoff = 1.0
  current_offset = last_offset

  while True:
    page_num += 1

    if len(records) >= max_records:
      print(f"Reached max_records limit ({max_records})")
      break

    params = [f"count={page_size}"]
    if current_offset:
      params.append(f"offset={current_offset}")

    url = f"{endpoint}?{'&'.join(params)}"

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

      if response.status == 429:
        retry_after = int(response.headers.get('Retry-After', str(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

      if response.status == 401:
        print("Token expired (401). Refreshing bearer token...")
        token = get_bearer_token()
        headers['Authorization'] = f'Bearer {token}'
        continue

      if response.status != 200:
        print(f"HTTP Error: {response.status}")
        response_text = response.data.decode('utf-8')
        print(f"Response body: {response_text}")
        return records, current_offset

      data = json.loads(response.data.decode('utf-8'))

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

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

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

      # Extract next offset from Link header for pagination
      link_header = response.headers.get('Link', '')
      next_offset = extract_next_offset(link_header)

      if next_offset:
        current_offset = next_offset
      else:
        print("No more pages (no next link)")
        break

    except Exception as e:
      print(f"Error fetching audit events: {e}")
      return records, current_offset

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

def extract_next_offset(link_header):
  if not link_header:
    return None

  parts = link_header.split(',')
  for part in parts:
    if 'rel="next"' in part:
      url_part = part.split(';')[0].strip().strip('<>')
      if 'offset=' in url_part:
        for param in url_part.split('?')[-1].split('&'):
          if param.startswith('offset='):
            return param.split('=', 1)[1]
  return None

def load_state(bucket):
  try:
    blob = bucket.blob(STATE_KEY)
    if blob.exists():
      return json.loads(blob.download_as_text())
  except Exception as e:
    print(f"Warning: Could not load state: {e}")
  return {}

def save_state(bucket, last_offset):
  try:
    state = {
      'last_offset': last_offset,
      'last_run': datetime.now(timezone.utc).isoformat()
    }
    blob = bucket.blob(STATE_KEY)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_offset={last_offset}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")

    • requirements.txt:

      functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. 点击部署以保存和部署函数。

  4. 等待部署完成（2-3 分钟）。

创建 Cloud Scheduler 作业

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

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

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

  4. 点击创建。

时间表频率选项

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

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

测试集成

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

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

    Starting audit log collection (last_offset=None)
Successfully obtained ASA bearer token
Page 1: Retrieved X events
Wrote X records to gs://okta-asa-audit-logs/okta-asa-audit/okta_asa_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. 前往 Cloud Storage > 存储桶。

  9. 点击 okta-asa-audit-logs。

  10. 转到 okta-asa-audit/ 文件夹。

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

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

  • HTTP 401：验证 ASA_KEY_ID 和 ASA_KEY_SECRET 环境变量是否正确
  • HTTP 403：验证服务用户是否具有 reporting_user 或 access_admin 角色
  • HTTP 429：速率限制 - 函数将自动重试，并采用指数退避算法
  • 缺少环境变量：验证是否已在 Cloud Run 函数配置中设置所有必需的变量

检索 Google SecOps 服务账号

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

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

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

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

  10. 点击下一步。

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

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

      gs://okta-asa-audit-logs/okta-asa-audit/

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

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

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

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

    • 资源命名空间：资源命名空间

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

  12. 点击下一步。

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

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

  1. 前往 Cloud Storage > 存储桶。
  2. 点击 okta-asa-audit-logs。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：粘贴 Google SecOps 服务账号邮箱
    • 分配角色：选择 Storage Object Viewer

  6. 点击保存。

UDM 映射表

日志字段 UDM 映射 逻辑
details.actor.details.email principal.user.email_addresses 直接复制值
details.actor.details.first_name principal.user.first_name 直接复制的值
details.actor.details.full_name principal.user.user_display_name 直接复制的值
details.actor.details.last_name principal.user.last_name 直接复制值
details.actor.details.user_type principal.user.attribute.roles 直接复制值
details.actor.id principal.user.employee_id 直接复制的值
details.actor.name principal.user.userid 如果 details.actor.name 不为空，则为该值；否则为 details.unix_user_name
details.unix_user_name principal.user.userid
details.actor.status principal.user.attribute.labels 合并为键为“actor_status”的标签
details.actor.team_name principal.user.department 直接复制的值
details.actor.user_type principal.user.attribute.labels 合并为键为“actor_user_type”的标签
details.client.description principal.application 直接复制值
details.client.encrypted principal.asset.attribute.labels 转换为字符串，合并为键为“details_client_encrypted”的标签
details.client.hostname principal.hostname、principal.asset.hostname 直接复制的值
details.client.id principal.user.product_object_id 直接复制的值
details.client.os principal.platform 如果匹配 (?i)Win|win，则设置为 WINDOWS；如果匹配 (?i)Lin|lin，则设置为 LINUX；如果匹配 (?i)Mac|ios|mac，则设置为 MAC；如果匹配 (?i)Android|android，则设置为 ANDROID
details.client.os principal.platform_version 使用 grok 从操作系统字符串中提取了 major.minor.patch
details.client.state principal.resource.attribute.labels 已合并为键为“client_state”的标签
details.client.user_name target.user.userid 直接复制值
details.client_ip principal.ip、principal.asset.ip 使用 grok 模式提取的 IP
details.from_address principal.ip、principal.asset.ip 使用 grok 模式提取的 IP
details.project.create_server_users target.resource.attribute.labels 转换为字符串，作为键为“project_create_server_users”的标签进行合并
details.project.force_shared_ssh_users target.resource.attribute.labels 转换为字符串，并与键“project_force_shared_ssh_users”合并为标签
details.project.forward_traffic target.resource.attribute.labels 转换为字符串，合并为键为“project_forward_traffic”的标签
details.project.id target.resource.id 直接复制值
details.project.name target.resource.name 直接复制的值
details.project.next_unix_gid target.resource.attribute.labels 已转换为字符串，并作为键为“next_unix_gid”的标签合并
details.project.next_unix_uid target.resource.attribute.labels 转换为字符串，合并为带有键“next_unix_uid”的标签
details.project.rdp_session_recording target.resource.attribute.labels 转换为字符串，合并为键为“project_rdp_session_recording”的标签
details.project.require_preauth_for_creds target.resource.attribute.labels 已转换为字符串，并作为键为“project_require_preauth_for_creds”的标签合并
details.project.ssh_certificate_type target.resource.attribute.labels 已合并为标签，密钥为“project_ssh_certificate_type”
details.project.ssh_session_recording target.resource.attribute.labels 转换为字符串，并作为标签与键“project_ssh_session_recording”合并
details.project.team target.resource.attribute.labels 已合并为键为“project_team”的标签
details.server.alt_names target.hostname、target.asset.hostname、target.resource.attribute.labels 第一个 alt_name 设置为 target.hostname 和 target.asset.hostname，其他合并为键为“alt_name”的标签
details.server.hostname intermediary_value.hostname 直接复制的值
details.server.id intermediary_value.resource.attribute.labels 合并为键为“details_server_id”的标签
details.server.os intermediary_value.platform_version 直接复制值
details.server.os intermediary_value.resource.attribute.labels 合并为键为“details_server_os”的标签
details.server.os_type intermediary_value.platform 如果匹配 (?i)linux，则设置为 LINUX；如果匹配 (?i)windows，则设置为 WINDOWS；如果匹配 (?i)mac，则设置为 MAC
details.server.services intermediary_value.resource.attribute.labels 合并为键为“server_services {index}”的标签
details.server.source intermediary_value.resource.attribute.labels 已合并为键为“server_source”的标签
details.server.source_details.cloud_account intermediary_value.resource.attribute.labels 已合并为键为“server_source_details_cloud_account”的标签
details.server.source_details.cloud_provider intermediary_value.resource.attribute.labels 合并为键为“server_source_details_cloud_provider”的标签
details.server.source_details.instance_id intermediary_value.resource.attribute.labels 合并为键为“server_source_details_instance_id”的标签
details.server.state target.asset.attribute.labels 转换为字符串，作为键为“server_state”的标签合并
details.server.team_name intermediary_value.user.department 直接复制的值
details.server_hostnames target.hostname、target.resource.attribute.labels 第一个主机名设置为 target.hostname，其他主机名合并为键为“server_hostname {index}”的标签
details.server_id target.resource.attribute.labels 合并为键为“details_server_id”的标签
details.server_os target.platform_version 直接复制的值
details.server_os_type target.platform 如果匹配 (?i)linux，则设置为 LINUX；如果匹配 (?i)windows，则设置为 WINDOWS；如果匹配 (?i)mac，则设置为 MAC
details.server_state target.resource.attribute.labels 合并为键为“details_server_state”的标签
details.session_type extensions.auth.auth_details 直接复制值
details.session_type security_result.detection_fields 已合并为键为“session_type”的标签
details.servers 中介 已处理并合并为中间对象
details.ssh_algorithm network.tls.cipher 直接复制值
details.ssh_key_fingerprint target.asset.attribute.labels 已合并为键为“ssh_key_fingerprint”的标签
details.ssh_public_key security_result.detection_fields 已合并为键为“ssh_public_key”的标签
details.target_server target.application 直接复制值
details.team_id principal.resource.id 直接复制的值
details.team_id target.resource.id 直接复制值
details.team_name additional.fields 已合并为键为“team_name”的标签
details.trace_id network.session_id 直接复制值
details.type metadata.product_event_type 直接复制值
details.username target.user.user_display_name 直接复制的值
details.via network.session_id 直接复制值
details_server intermediary_value 已处理并合并为 intermediary_value 对象
id additional.fields 已合并为键为“id”的标签
时间戳 metadata.event_timestamp 已使用 ISO8601 格式转换
extensions.auth.type extensions.auth.type 如果 metadata.event_type 为 USER_LOGIN，则设置为“SSO”
intermediary_value 中介 如果不为空，则从 intermediary_value 合并
has_principal_user metadata.event_type 如果具有 has_principal_user 和 has_target_user 或具有 has_principal_user 和 has_target 且 details_type 与登录匹配，则设置为 USER_LOGIN；否则，如果具有 has_principal_user，则设置为 USER_UNCATEGORIZED；否则，如果具有 has_principal，则设置为 STATUS_UPDATE
has_target_user metadata.event_type
metadata.product_name metadata.product_name 设置为“OKTA_SCALEFT”
metadata.vendor_name metadata.vendor_name 设置为“OKTA_SCALEFT”
desc event.idm.read_only_udm.metadata.description 从更新日志映射
asset_id event.idm.read_only_udm.target.asset.asset_id 从更新日志映射
details.hostname event.idm.read_only_udm.principal.hostname和event.idm.read_only_udm.principal.asset.hostname之间 从更新日志映射
details.actor event.idm.read_only_udm.principal.user.userid 从更新日志映射
details.client event.idm.read_only_udm.principal.hostname和event.idm.read_only_udm.principal.asset.hostname之间 从更新日志映射
details.username event.idm.read_only_udm.target.user.user_display_name 从更新日志映射
details.server.server_accounts.username event.idm.read_only_udm.target.user.user_display_name 从更新日志映射
details.project event.idm.read_only_udm.target.resource.attribute.cloud.project.name 从更新日志映射
details.server.server_accounts event.idm.read_only_udm.target.asset.attribute.labels 从更新日志映射
"id" event.idm.read_only_udm.additional.fields 从更新日志映射
"details.from_address" event.idm.read_only_udm.principal.ip 从更新日志映射
"details.client_ip" event.idm.read_only_udm.principal.ip 从更新日志映射
"details.from_address" event.idm.read_only_udm.principal.asset.ip 从更新日志映射
"details.client_ip" event.idm.read_only_udm.principal.asset.ip 从更新日志映射
"details.type" event.idm.read_only_udm.metadata.product_event_type 从更新日志映射
"details.actor.status" event.idm.read_only_udm.principal.user.attribute.labels 从更新日志映射
"details.actor.user_type" event.idm.read_only_udm.principal.user.attribute.labels 从更新日志映射
"details.session_type" event.idm.read_only_udm.extensions.auth.auth_details 从更新日志映射
"details.actor.name" event.idm.read_only_udm.principal.user.userid 从更新日志映射
"details.unix_user_name" event.idm.read_only_udm.principal.user.userid 从更新日志映射
"details.actor.details.full_name" event.idm.read_only_udm.principal.user.user_display_name 从更新日志映射
"details.actor.details.first_name" event.idm.read_only_udm.principal.user.first_name 从更新日志映射
"details.actor.details.last_name" event.idm.read_only_udm.principal.user.last_name 从更新日志映射
"details.actor.id" event.idm.read_only_udm.principal.user.employee_id 从更新日志映射
"details.actor.team_name" event.idm.read_only_udm.principal.user.department 从更新日志映射
"details.client.hostname" event.idm.read_only_udm.principal.hostname 从更新日志映射
"details.client.hostname" event.idm.read_only_udm.principal.asset.hostname 从更新日志映射
"details.client.id" event.idm.read_only_udm.principal.user.product_object_id 从更新日志映射
"details.client.description" event.idm.read_only_udm.principal.application 从更新日志映射
"details.client.os" event.idm.read_only_udm.principal.platform 从更新日志映射
"details_client_os" event.idm.read_only_udm.principal.platform_version 从更新日志映射
"details.actor.details.email" event.idm.read_only_udm.principal.user.email_addresses 从更新日志映射
"details.team_id" event.idm.read_only_udm.target.resource.id 从更新日志映射
"details.project.id" event.idm.read_only_udm.target.resource.id 从更新日志映射
"details.team_name" event.idm.read_only_udm.additional.fields 从更新日志映射
"details.client.state" event.idm.read_only_udm.principal.resource.attribute.labels 从更新日志映射
"details.target_server" event.idm.read_only_udm.target.application 从更新日志映射
"details.via" event.idm.read_only_udm.network.session_id 从更新日志映射
"details.trace_id" event.idm.read_only_udm.network.session_id 从更新日志映射
"details_session_type" event.idm.read_only_udm.security_result.detection_fields 从更新日志映射
"details.ssh_public_key" event.idm.read_only_udm.security_result.detection_fields 从更新日志映射
"details.actor.details.user_type" event.idm.read_only_udm.principal.user.attribute.roles 从更新日志映射
"details.project.name" event.idm.read_only_udm.target.resource.name 从更新日志映射
"details.server.id" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.server.state" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.team" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.create_server_users" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.force_shared_ssh_users" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.forward_traffic" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.next_unix_gid" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.next_unix_uid" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.rdp_session_recording" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.require_preauth_for_creds" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.ssh_certificate_type" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.project.ssh_session_recording" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"serverhostname" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"alt_name" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.ssh_key_fingerprint" event.idm.read_only_udm.target.resource.attribute.labels 从更新日志映射
"details.server.os" event.idm.read_only_udm.target.platform_version 从更新日志映射
"detailsserver.os" event.idm.read_only_udm.target.platform_version 从更新日志映射
"details.server.os_type" event.idm.read_only_udm.target.platform 从更新日志映射
"detailsserver.os_type" event.idm.read_only_udm.target.platform 从更新日志映射
"details_team_id" event.idm.read_only_udm.principal.resource.id 从更新日志映射
"details.client.encrypted" event.idm.read_only_udm.principal.asset.attribute.labels 从更新日志映射
"details.client.user_name" event.idm.read_only_udm.target.user.userid 从更新日志映射
"timestamp" event.idm.read_only_udm.metadata.event_timestamp 从更新日志映射
"serverhostname" event.idm.read_only_udm.target.hostname 从更新日志映射
"alt_name" event.idm.read_only_udm.target.hostname 从更新日志映射
"alt_name" event.idm.read_only_udm.target.asset.hostname 从更新日志映射
"intermediary" event.idm.read_only_udm.intermediary 从更新日志映射
"intermediary_value" event.idm.read_only_udm.intermediary 从更新日志映射
"detailsserver.sftd_version" event.idm.read_only_udm.metadata.product_version 从更新日志映射
"details.server.sftd_version" event.idm.read_only_udm.metadata.product_version 从更新日志映射
"detailsserver.state" event.idm.read_only_udm.target.asset.attribute.labels 从更新日志映射
"detailsserver.managed" event.idm.read_only_udm.target.asset.attribute.labels 从更新日志映射
"detailsserver.ssh_key_fingerprint" event.idm.read_only_udm.target.asset.attribute.labels 从更新日志映射
"details.username" event.idm.read_only_udm.target.user.user_display_name 从更新日志映射
"details.ssh_algorithm" event.idm.read_only_udm.network.tls.cipher 从更新日志映射

更新日志

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

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