收集 Zoom 操作日志
本文档介绍了如何使用 Google Cloud Storage 将 Zoom 操作日志提取到 Google Security Operations。解析器会将原始日志转换为统一数据模型 (UDM)。它从原始日志消息中提取字段,执行数据清理和归一化,并将提取的信息映射到相应的 UDM 字段,最终丰富数据以便在 SIEM 系统中进行分析和关联。
准备工作
确保您满足以下前提条件:
- Google SecOps 实例
- 已启用 Cloud Storage API 的 GCP 项目
- 创建和管理 GCS 存储分区的权限
- 管理 GCS 存储分区的 IAM 政策的权限
- 创建 Cloud Run 函数、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 对 Zoom 的特权访问权限
收集 Zoom 操作日志的前提条件
- 登录 Zoom App Marketplace。
- 依次前往开发 > 构建应用 > 服务器到服务器 OAuth。
- 创建应用并添加以下范围:
report:read:operation_logs:admin(或report:read:admin)。 在应用凭据中,复制以下详细信息并将其保存在安全的位置:
- 账号 ID
- Client-ID
- 客户端密钥 (Client Secret)
验证权限
如需验证账号是否具有所需权限,请执行以下操作:
- 登录您的 Zoom 账号。
- 依次前往“管理员”图标 >“账号管理”>“账号资料”。
- 如果您可以访问账号设置并查看操作日志,则表示您拥有所需的权限。
- 如果您无法使用这些选项,请与您的 Zoom 管理员联系,让其授予您必要的权限。
测试 API 访问权限
在继续进行集成之前,请先测试您的凭据:
# Replace with your actual credentials ZOOM_ACCOUNT_ID="<your-account-id>" ZOOM_CLIENT_ID="<your-client-id>" ZOOM_CLIENT_SECRET="<your-client-secret>" # Get OAuth token TOKEN=$(curl -s -X POST "https://zoom.us/oauth/token?grant_type=account_credentials&account_id=${ZOOM_ACCOUNT_ID}" \ -u "${ZOOM_CLIENT_ID}:${ZOOM_CLIENT_SECRET}" \ | grep -o '"access_token":"[^"]*"' | cut -d'"' -f4) # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" \ "https://api.zoom.us/v2/report/operationlogs?from=$(date -u -d '1 day ago' +%Y-%m-%d)&to=$(date -u +%Y-%m-%d)&page_size=10"
创建 Google Cloud Storage 存储分区
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中,依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息:
设置 值 为存储分区命名 输入一个全局唯一的名称(例如 zoom-operation-logs)位置类型 根据您的需求进行选择(区域级、双区域级、多区域级) 位置 选择相应位置(例如 us-central1)存储类别 标准(建议用于经常访问的日志) 访问权限控制 统一(推荐) 保护工具 可选:启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号,该账号具有向 GCS 存储分区写入内容以及被 Pub/Sub 调用的权限。
创建服务账号
- 在 GCP 控制台中,依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息:
- 服务账号名称:输入
zoom-operationlogs-sa。 - 服务账号说明:输入
Service account for Cloud Run function to collect Zoom operation logs。
- 服务账号名称:输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分中,添加以下角色:
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
必须拥有这些角色,才能:
- Storage Object Admin:将日志写入 GCS 存储分区并管理状态文件
- Storage Object Admin:将日志写入 GCS 存储分区并管理状态文件
- Cloud Run Invoker:允许 Pub/Sub 调用函数
- Cloud Functions Invoker:允许调用函数
授予对 GCS 存储分区的 IAM 权限
向服务账号授予对 GCS 存储分区的写入权限:
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称(例如
zoom-operation-logs)。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:输入服务账号电子邮件地址(例如
zoom-operationlogs-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 分配角色:选择 Storage Object Admin。
- 添加主账号:输入服务账号电子邮件地址(例如
- 点击保存。
创建发布/订阅主题
创建一个 Pub/Sub 主题,供 Cloud Scheduler 发布消息,并供 Cloud Run 函数订阅。
- 在 GCP 控制台中,前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息:
- 主题 ID:输入
zoom-operationlogs-trigger。 - 将其他设置保留为默认值。
- 主题 ID:输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发,用于从 Zoom API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中,前往 Cloud Run。
- 点击创建服务。
- 选择函数(使用内嵌编辑器创建函数)。
在配置部分中,提供以下配置详细信息:
设置 值 Service 名称 zoom-operationlogs-to-gcs区域 选择与您的 GCS 存储分区匹配的区域(例如 us-central1)运行时 选择 Python 3.12 或更高版本 在触发器(可选)部分中:
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中,选择
zoom-operationlogs-trigger。 - 点击保存。
在身份验证部分中:
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全标签页:
- 服务账号:选择
zoom-operationlogs-sa。
- 服务账号:选择
前往容器标签页:
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量:
变量名称 示例值 GCS_BUCKETzoom-operation-logsGCS_PREFIXzoom/operationlogs/STATE_KEYzoom/operationlogs/state.jsonZOOM_ACCOUNT_ID<your-zoom-account-id>ZOOM_CLIENT_ID<your-zoom-client-id>ZOOM_CLIENT_SECRET<your-zoom-client-secret>PAGE_SIZE300TIMEOUT30在变量和 Secret 部分中,滚动到请求:
- 请求超时:输入
600秒(10 分钟)。
- 请求超时:输入
前往设置标签页:
- 在资源部分中:
- 内存:选择 512 MiB 或更高值。
- CPU:选择 1。
- 在资源部分中:
在修订版本扩缩部分中:
- 实例数下限:输入
0。 - 实例数上限:输入
100(或根据预期负载进行调整)。
- 实例数下限:输入
点击创建。
等待服务创建完成(1-2 分钟)。
创建服务后,系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点中输入 main
在内嵌代码编辑器中,创建两个文件:
- 第一个文件:main.py::
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, date, timedelta, timezone import base64 import uuid import gzip import io # 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', 'zoom/operationlogs/') STATE_KEY = os.environ.get('STATE_KEY', 'zoom/operationlogs/state.json') ZOOM_ACCOUNT_ID = os.environ.get('ZOOM_ACCOUNT_ID') ZOOM_CLIENT_ID = os.environ.get('ZOOM_CLIENT_ID') ZOOM_CLIENT_SECRET = os.environ.get('ZOOM_CLIENT_SECRET') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '300')) TIMEOUT = int(os.environ.get('TIMEOUT', '30')) TOKEN_URL = "https://zoom.us/oauth/token" REPORT_URL = "https://api.zoom.us/v2/report/operationlogs" @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Zoom operation logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ZOOM_ACCOUNT_ID, ZOOM_CLIENT_ID, ZOOM_CLIENT_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Get OAuth token token = get_token() # Load state state = load_state(bucket, STATE_KEY) cursor_date = state.get('cursor_date', date.today().isoformat()) print(f'Processing logs for date: {cursor_date}') # Fetch logs from_date = cursor_date to_date = cursor_date total_written = 0 next_token = state.get('next_page_token') while True: page = fetch_page(token, from_date, to_date, next_token) items = page.get('operation_logs', []) or [] if items: write_chunk(bucket, items, datetime.now(timezone.utc)) total_written += len(items) next_token = page.get('next_page_token') if not next_token: break # Advance to next day if we've finished this date today = date.today().isoformat() if cursor_date < today: nxt = (datetime.fromisoformat(cursor_date) + timedelta(days=1)).date().isoformat() state['cursor_date'] = nxt state['next_page_token'] = None else: # stay on today; continue later with next_page_token=None state['next_page_token'] = None save_state(bucket, STATE_KEY, state) print(f'Successfully processed {total_written} logs for {from_date}') except Exception as e: print(f'Error processing logs: {str(e)}') raise def get_token(): """Get OAuth 2.0 access token from Zoom.""" params = f"grant_type=account_credentials&account_id={ZOOM_ACCOUNT_ID}" basic = base64.b64encode(f"{ZOOM_CLIENT_ID}:{ZOOM_CLIENT_SECRET}".encode('utf-8')).decode('utf-8') headers = { 'Authorization': f'Basic {basic}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json', 'Host': 'zoom.us' } response = http.request( 'POST', TOKEN_URL, body=params, headers=headers, timeout=TIMEOUT ) if response.status != 200: print(f'Token request failed: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') raise Exception(f'Failed to get OAuth token: {response.status}') body = json.loads(response.data.decode('utf-8')) return body['access_token'] def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f'Warning: Could not load state: {str(e)}') # Initial state: start today today = date.today().isoformat() return {'cursor_date': today, 'next_page_token': None} def save_state(bucket, key, state): """Save state to GCS.""" try: state['updated_at'] = datetime.now(timezone.utc).isoformat() blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}') def write_chunk(bucket, items, ts): """Write log chunk to GCS.""" key = f"{GCS_PREFIX}{ts:%Y/%m/%d}/zoom-operationlogs-{uuid.uuid4()}.json.gz" buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode='w') as gz: for rec in items: gz.write((json.dumps(rec) + '\n').encode('utf-8')) buf.seek(0) blob = bucket.blob(key) blob.upload_from_file(buf, content_type='application/gzip') print(f'Wrote {len(items)} logs to {key}') return key def fetch_page(token, from_date, to_date, next_page_token): """Fetch a page of logs from Zoom API.""" params = { 'from': from_date, 'to': to_date, 'page_size': str(PAGE_SIZE) } if next_page_token: params['next_page_token'] = next_page_token # Build query string query_string = '&'.join([f"{k}={v}" for k, v in params.items()]) url = f"{REPORT_URL}?{query_string}" headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json' } response = http.request( 'GET', url, headers=headers, timeout=TIMEOUT ) if response.status != 200: print(f'API request failed: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') raise Exception(f'Failed to fetch logs: {response.status}') return json.loads(response.data.decode('utf-8'))- 第二个文件:requirements.txt::
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署该函数。
等待部署完成(2-3 分钟)。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息,从而触发 Cloud Run 函数。
- 在 GCP Console 中,前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息:
设置 值 名称 zoom-operationlogs-schedule-15min区域 选择与 Cloud Run 函数相同的区域 频率 */15 * * * *(每 15 分钟)时区 选择时区(建议选择世界协调时间 [UTC]) 目标类型 Pub/Sub 主题 选择 zoom-operationlogs-trigger消息正文 {}(空 JSON 对象)点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次:
频率 Cron 表达式 使用场景 每隔 5 分钟 */5 * * * *高容量、低延迟 每隔 15 分钟 */15 * * * *标准(推荐) 每小时 0 * * * *搜索量较低 每 6 小时 0 */6 * * *批处理
测试集成
- 在 Cloud Scheduler 控制台中,找到您的作业(例如
zoom-operationlogs-schedule-15min)。 - 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击函数名称 (
zoom-operationlogs-to-gcs)。 - 点击日志标签页。
验证函数是否已成功执行。查找以下项:
Processing logs for date: YYYY-MM-DD Page 1: Retrieved X events Wrote X records to zoom/operationlogs/YYYY/MM/DD/zoom-operationlogs-UUID.json.gz Successfully processed X logs for YYYY-MM-DD前往 Cloud Storage > 存储分区。
点击您的存储分区名称 (
zoom-operation-logs)。前往前缀文件夹 (
zoom/operationlogs/)。验证是否已创建具有当前时间戳的新
.json.gz文件。
如果您在日志中看到错误,请执行以下操作:
- HTTP 401:检查环境变量中的 Zoom API 凭据
- HTTP 403:验证 Zoom 应用是否具有
report:read:operation_logs:admin范围 - 缺少环境变量:检查是否已在 Cloud Run 函数配置中设置所有必需的变量
检索 Google SecOps 服务账号
Google SecOps 使用唯一的服务账号从您的 GCS 存储分区中读取数据。您必须授予此服务账号对您的存储分区的访问权限。
获取服务账号电子邮件地址
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中,输入 Feed 的名称(例如
Zoom Operation Logs)。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Zoom Operation Logs 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址,以便在下一步中使用。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要对您的 GCS 存储分区具有 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称(例如
zoom-operation-logs)。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色:选择 Storage Object Viewer。
点击保存。
在 Google SecOps 中配置 Feed 以提取 Zoom 操作日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中,输入 Feed 的名称(例如
Zoom Operation Logs)。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Zoom Operation Logs 作为日志类型。
- 点击下一步。
为以下输入参数指定值:
存储分区网址:输入带有前缀路径的 GCS 存储分区 URI:
gs://zoom-operation-logs/zoom/operationlogs/将
zoom-operation-logs:您的 GCS 存储分区名称。zoom/operationlogs/:存储日志的前缀路径。
来源删除选项:根据您的偏好选择删除选项:
- 永不:永不删除转移后的任何文件(建议用于测试)。
- 删除已转移的文件:在成功转移后删除文件。
- 删除已转移的文件和空目录:在成功转移后删除文件和空目录。
文件存在时间上限:包含在过去指定天数内修改的文件。默认值为 180 天。
资产命名空间:资产命名空间。
注入标签:要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置,然后点击提交。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
| action | metadata.product_event_type | 原始日志字段“action”映射到此 UDM 字段。 |
| category_type | additional.fields.key | 原始日志字段“category_type”映射到此 UDM 字段。 |
| category_type | additional.fields.value.string_value | 原始日志字段“category_type”映射到此 UDM 字段。 |
| 省 | target.user.department | 原始日志字段“Department”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 说明 | target.user.role_description | 原始日志字段“Description”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 显示名称 | target.user.user_display_name | 原始日志字段“显示名称”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 电子邮件地址 | target.user.email_addresses | 原始日志字段“电子邮件地址”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 名字 | target.user.first_name | 原始日志字段“名字”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 职位 | target.user.title | 原始日志字段“职位”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 姓氏 | target.user.last_name | 原始日志字段“Last Name”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 位置 | target.location.name | 原始日志字段“位置”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| operation_detail | metadata.description | 原始日志字段“operation_detail”会映射到此 UDM 字段。 |
| operator | principal.user.email_addresses | 如果原始日志字段“operator”与电子邮件正则表达式匹配,则会映射到此 UDM 字段。 |
| operator | principal.user.userid | 如果原始日志字段“operator”与电子邮件正则表达式不匹配,则会映射到此 UDM 字段。 |
| 房间名称 | target.user.attribute.labels.value | 原始日志字段“会议室名称”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 角色名称 | target.user.attribute.roles.name | 原始日志字段“角色名称”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 时间 | metadata.event_timestamp.seconds | 原始日志字段“time”经过解析后映射到此 UDM 字段。 |
| 类型 | target.user.attribute.labels.value | 原始日志字段“Type”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 用户角色 | target.user.attribute.roles.name | 原始日志字段“用户角色”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| 用户类型 | target.user.attribute.labels.value | 原始日志字段“用户类型”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| metadata.log_type | 系统会为相应 UDM 字段分配值“ZOOM_OPERATION_LOGS”。 | |
| metadata.vendor_name | 系统会为此 UDM 字段分配值“ZOOM”。 | |
| metadata.product_name | 系统会为相应 UDM 字段分配值“ZOOM_OPERATION_LOGS”。 | |
| metadata.event_type | 该值根据以下逻辑确定:1. 如果“event_type”字段不为空,则使用其值。1. 如果“operator”“email”或“email2”字段不为空,则该值会设置为“USER_UNCATEGORIZED”。1. 否则,该值设置为“GENERIC_EVENT”。 | |
| json_data | about.user.attribute.labels.value | 原始日志字段“json_data”(从“operation_detail”字段中提取)被解析为 JSON。已解析的 JSON 数组中每个元素的“assistant”和“options”字段会映射到 UDM 中“labels”数组的“value”字段。 |
| json_data | about.user.userid | 原始日志字段“json_data”(从“operation_detail”字段中提取)被解析为 JSON。已解析的 JSON 数组中每个元素(第一个元素除外)的“userId”字段会映射到 UDM 中“about.user”对象的“userid”字段。 |
| json_data | target.user.attribute.labels.value | 原始日志字段“json_data”(从“operation_detail”字段中提取)被解析为 JSON。解析后的 JSON 数组的第一个元素中的“assistant”和“options”字段会映射到 UDM 中“labels”数组的“value”字段。 |
| json_data | target.user.userid | 原始日志字段“json_data”(从“operation_detail”字段中提取)被解析为 JSON。已解析的 JSON 数组中第一个元素的“userId”字段会映射到 UDM 中“target.user”对象的“userid”字段。 |
| 电子邮件 | target.user.email_addresses | 原始日志字段“email”(从“operation_detail”字段中提取)会映射到此 UDM 字段。 |
| email2 | target.user.email_addresses | 原始日志字段“email2”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
| 角色 | target.user.attribute.roles.name | 原始日志字段“role”(从“operation_detail”字段中提取)映射到此 UDM 字段。 |
需要更多帮助?获得社区成员和 Google SecOps 专业人士的解答。