收集 Okta Advanced Server Access 日志

支持的平台：

Google SecOps SIEM

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

本文档介绍了如何使用 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. 登录 Advanced Server Access 信息中心。
2. 转到用户。
3. 选择服务用户标签页。
4. 点击创建服务用户。

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

   注意： 系统会根据您提供的用户名自动创建相应的 Linux 和 Windows 用户名。

6. 点击创建服务用户。

生成 API 密钥

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

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

   • API 密钥 ID：用于标识 API 密钥的 UUID（例如 6052868b-1b04-4a14-8288-e6496d7f2f4b）
   • API 密钥密文：用于对 API 请求进行身份验证的 base64 编码密文
   注意： 关闭此窗口后，您将无法检索 API 密钥 Secret。如果您丢失了此信息，则必须生成新的 API 密钥。

3. 点击完成。

将服务用户分配给具有相应权限的群组

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

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

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

   • reporting_user：对所有端点具有只读权限（建议用于日志提取）
   • access_admin：完整的管理员权限（授予所有 CRUD 权限）
   注意： 如需实现最小权限访问，请使用 reporting_user 角色。此角色提供对审核事件的只读访问权限，但不授予管理权限。

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

验证权限

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

1. 在“高级服务器访问权限”信息中心内，前往群组。
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)。
   注意： 调用函数时，Pub/Sub 会自动处理身份验证。

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	高级服务器访问权限团队名称
   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. 在变量和 Secret 部分中，向下滚动到请求：

    • 请求超时：输入 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 分钟）。

   注意： Pub/Sub 触发器配置会自动创建必要的订阅和权限。

创建 Cloud Scheduler 作业

1. 在 GCP Console 中，前往 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. 系统会显示一个唯一的服务账号电子邮件地址。例如：

   [e-mail]
   

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

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

10. 点击下一步。

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

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

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

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

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

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

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

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

    • 文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 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 专业人士的解答。