收集 Zoom 作業記錄
本文說明如何使用 Google Cloud Storage,將 Zoom 作業記錄擷取至 Google Security Operations。剖析器會將原始記錄轉換為統合式資料模型 (UDM)。這項功能會從原始記錄訊息中擷取欄位、執行資料清理和正規化作業,並將擷取的資訊對應至相應的 UDM 欄位,最終擴充資料,以便在 SIEM 系統中進行分析和關聯。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- Zoom 的特殊存取權
收集 Zoom 作業記錄檔的必要條件
- 登入 Zoom 應用程式市集。
- 依序前往「Develop」>「Build App」>「Server-to-Server OAuth」。
- 建立應用程式並新增下列範圍:
report:read:operation_logs:admin(或report:read:admin)。 在「應用程式憑證」中,複製下列詳細資料並儲存在安全位置:
- 帳戶 ID
- 用戶端 ID
- 用戶端密碼
驗證權限
如要確認帳戶是否具備必要權限,請按照下列步驟操作:
- 登入 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」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 zoom-operation-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶,並由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 主控台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下 [Create Service Account] (建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
zoom-operationlogs-sa。 - 服務帳戶說明:輸入
Service account for Cloud Run function to collect Zoom operation logs。
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
- Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
zoom-operation-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
zoom-operationlogs-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
zoom-operationlogs-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Zoom API 擷取記錄並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 zoom-operationlogs-to-gcs區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中,選擇
zoom-operationlogs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁:
- 服務帳戶:選取
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。 - 「Maximum number of instances」(執行個體數量上限):輸入
100(或根據預期負載調整)。
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「Function entry point」(函式進入點) 中輸入 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 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 zoom-operationlogs-schedule-15min區域 選取與 Cloud Run 函式相同的區域 頻率 */15 * * * *(每 15 分鐘)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取「 zoom-operationlogs-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *高容量、低延遲 每 15 分鐘檢查一次 */15 * * * *標準 (建議採用) 每小時 0 * * * *低量 每 6 小時 0 */6 * * *批次處理
測試整合項目
- 在 Cloud Scheduler 控制台中,找出您的工作 (例如
zoom-operationlogs-schedule-15min)。 - 按一下「強制執行」,手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱 (
zoom-operationlogs-to-gcs)。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。請找出以下項目:
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」>「Buckets」。
按一下 bucket 名稱 (
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 bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Zoom Operation Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Zoom Operation Logs」(Zoom 作業記錄) 做為「Log type」(記錄類型)。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
zoom-operation-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定資訊提供,擷取 Zoom 作業記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Zoom Operation Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Zoom Operation Logs」(Zoom 作業記錄) 做為「Log type」(記錄類型)。
- 點選 [下一步]。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://zoom-operation-logs/zoom/operationlogs/取代:
zoom-operation-logs:您的 GCS bucket 名稱。zoom/operationlogs/:儲存記錄的路徑前置字元。
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
- 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| 動作 | 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 | 原始記錄欄位「Job Title」(從「operation_detail」欄位擷取) 會對應至這個 UDM 欄位。 |
| 姓氏 | target.user.last_name | 原始記錄欄位「Last Name」(從「operation_detail」欄位擷取) 會對應至這個 UDM 欄位。 |
| 位置 | target.location.name | 原始記錄欄位「Location」(從「operation_detail」欄位擷取) 會對應至這個 UDM 欄位。 |
| operation_detail | metadata.description | 原始記錄欄位「operation_detail」會對應至這個 UDM 欄位。 |
| 運算子 | principal.user.email_addresses | 如果原始記錄欄位「operator」符合電子郵件規則運算式,就會對應至這個 UDM 欄位。 |
| 運算子 | principal.user.userid | 如果原始記錄欄位「operator」不符合電子郵件正則運算式,就會對應至這個 UDM 欄位。 |
| 房間名稱 | target.user.attribute.labels.value | 原始記錄欄位「Room Name」(從「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 | 「ZOOM_OPERATION_LOGS」值會指派給這個 UDM 欄位。 | |
| metadata.vendor_name | 系統會將「ZOOM」值指派給這個 UDM 欄位。 | |
| metadata.product_name | 「ZOOM_OPERATION_LOGS」值會指派給這個 UDM 欄位。 | |
| 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 專業人員尋求答案。