收集 TeamViewer 記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，將 TeamViewer 記錄檔擷取至 Google Security Operations。剖析器會從 JSON 格式的記錄中擷取稽核事件。這項服務會逐一檢查活動詳細資料，將特定屬性對應至整合式資料模型 (UDM) 欄位、處理參與者和簡報者資訊，並根據使用者活動將活動分類。剖析器也會執行資料轉換，例如合併標籤，以及將時間戳記轉換為標準化格式。

事前準備

請確認您已完成下列事前準備事項：

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
TeamViewer 管理控制台的特殊權限
TeamViewer Business、Premium、Corporate 或 Tensor 授權 (存取 API 時必須具備)

建立 Google Cloud Storage 值區

前往 Google Cloud 控制台。
選取專案或建立新專案。
在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。

請提供下列設定詳細資料：

設定	值
為 bucket 命名	輸入全域不重複的名稱 (例如 `teamviewer-logs`)
位置類型	根據需求選擇 (區域、雙區域、多區域)
位置	選取地點 (例如 `us-central1`)
儲存空間級別	標準 (建議用於經常存取的記錄)
存取控管	統一 (建議)
保護工具	選用：啟用物件版本管理或保留政策

點選「建立」。

取得 TeamViewer 先決條件

前往 https://login.teamviewer.com/ 登入 TeamViewer 管理主控台。
按一下右上角的使用者圖示，然後選取「編輯個人資料」。
選取「應用程式」。
按一下「建立指令碼權杖」。
請提供下列設定詳細資料：
- 權杖名稱：輸入描述性名稱 (例如 Google SecOps Integration)。
- 權限：選取下列權限：
  - 「帳戶管理」>「查看帳戶資料」
  - 「工作階段管理」>「查看工作階段資料」
  - 連線報告 > 查看連線報告
點選「建立」。
複製並將產生的指令碼權杖儲存在安全的位置。

注意： 指令碼權杖只會顯示一次。如果遺失，請建立新權杖。指令碼權杖必須包含與 Connections.Read 範圍對應的連線報告權限，/api/v1/reports/connections 需要這項權限。
記錄 TeamViewer API 基礎網址：https://webapi.teamviewer.com/api/v1

驗證權限

如要確認帳戶是否具備必要權限，請按照下列步驟操作：

登入 TeamViewer 管理控制台。
依序前往「編輯商家檔案」>「應用程式」。
在清單中找出指令碼權杖。
確認已啟用「連線報告」>「查看連線報告」。
如果未啟用這項權限，請編輯權杖並新增必要權限。

測試 API 存取權

請先測試憑證，再繼續進行整合：

# Replace with your actual script token
SCRIPT_TOKEN="your-script-token"
API_BASE="https://webapi.teamviewer.com/api/v1"

# Test API access
curl -v -H "Authorization: Bearer ${SCRIPT_TOKEN}" \
  -H "Accept: application/json" \
  "${API_BASE}/reports/connections?from_date=2024-01-01T00:00:00Z&to_date=2024-01-01T01:00:00Z"

如果收到含有 JSON 資料的 200 回應，表示您的憑證設定正確無誤。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下 [Create Service Account] (建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：輸入 teamviewer-collector-sa。
- 服務帳戶說明：輸入 Service account for Cloud Run function to collect TeamViewer logs。
按一下「建立並繼續」。
在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
1. 按一下「選擇角色」。
2. 搜尋並選取「Storage 物件管理員」。
3. 點選「+ 新增其他角色」。
4. 搜尋並選取「Cloud Run Invoker」。
5. 點選「+ 新增其他角色」。
6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
按一下「繼續」。
按一下 [完成]。

這些角色是下列作業的必要條件： - Storage 物件管理員：將記錄寫入 GCS bucket，並管理狀態檔案 - Cloud Run 叫用者：允許 Pub/Sub 叫用函式 - Cloud Functions 叫用者：允許叫用函式

授予 GCS 值區的 IAM 權限

授予服務帳戶 GCS bucket 的寫入權限：

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (例如 teamviewer-logs)。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如 teamviewer-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
- 指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入 teamviewer-logs-trigger。
- 其他設定保留預設值。
點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 TeamViewer API 擷取記錄，並將記錄寫入 GCS。

前往 GCP Console 的「Cloud Run」。
按一下「Create service」(建立服務)。
選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：

設定值

服務名稱 teamviewer-logs-collector

區域選取與 GCS bucket 相符的區域 (例如 us-central1)

執行階段 選取「Python 3.12」以上版本
在「Trigger (optional)」(觸發條件 (選用)) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (teamviewer-logs-trigger)。
4. 按一下 [儲存]。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 檢查 Identity and Access Management (IAM)。
注意： 叫用函式時，Pub/Sub 會自動處理驗證。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (teamviewer-collector-sa)。

設定	值
服務名稱	`teamviewer-logs-collector`
區域	選取與 GCS bucket 相符的區域 (例如 `us-central1`)
執行階段	選取「Python 3.12」以上版本

前往「容器」分頁：

按一下「變數與密鑰」。
針對每個環境變數，按一下「+ 新增變數」：

變數名稱	範例值
`GCS_BUCKET`	`teamviewer-logs`
`GCS_PREFIX`	`teamviewer/audit/`
`STATE_KEY`	`teamviewer/audit/state.json`
`WINDOW_SECONDS`	`3600`
`HTTP_TIMEOUT`	`60`
`MAX_RETRIES`	`3`
`USER_AGENT`	`teamviewer-to-gcs/1.0`
`SCRIPT_TOKEN`	`your-script-token` (來自 TeamViewer 先決條件)
`API_BASE_URL`	`https://webapi.teamviewer.com/api/v1`

在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入 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 urllib.request
import urllib.parse
import urllib.error
from datetime import datetime, timezone
import time
import uuid

# Initialize Storage client
storage_client = storage.Client()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch TeamViewer audit logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'teamviewer/audit/')
    state_key = os.environ.get('STATE_KEY', 'teamviewer/audit/state.json')
    window_sec = int(os.environ.get('WINDOW_SECONDS', '3600'))
    http_timeout = int(os.environ.get('HTTP_TIMEOUT', '60'))
    max_retries = int(os.environ.get('MAX_RETRIES', '3'))
    user_agent = os.environ.get('USER_AGENT', 'teamviewer-to-gcs/1.0')

    # TeamViewer API credentials
    api_base_url = os.environ.get('API_BASE_URL')
    script_token = os.environ.get('SCRIPT_TOKEN')

    if not all([bucket_name, api_base_url, script_token]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        # Load state (last processed timestamp)
        state = load_state(bucket, state_key)
        now = time.time()
        from_ts = float(state.get('last_to_ts') or (now - window_sec))
        to_ts = now

        print(f'Fetching TeamViewer audit data from {iso_format(from_ts)} to {iso_format(to_ts)}')

        # Build audit API URL
        url = build_audit_url(api_base_url, from_ts, to_ts)

        print(f'Fetching TeamViewer audit data from: {url}')

        # Fetch audit data with retries and pagination
        all_records = []
        offset_id = None

        while True:
            blob_data, content_type, next_offset = fetch_audit_data(
                url, script_token, user_agent, http_timeout, max_retries, offset_id
            )

            # Validate JSON data
            try:
                audit_data = json.loads(blob_data)
                records = audit_data.get('records', [])
                all_records.extend(records)
                print(f"Retrieved {len(records)} audit records (total: {len(all_records)})")

                # Check for pagination
                if next_offset and len(records) == 1000:
                    offset_id = next_offset
                    print(f"Fetching next page with offset_id: {offset_id}")
                else:
                    break

            except json.JSONDecodeError as e:
                print(f"Warning: Invalid JSON received: {e}")
                break

        if all_records:
            # Write to GCS
            key = put_audit_data(bucket, prefix, json.dumps({'records': all_records}), 
                               'application/json', from_ts, to_ts)
            print(f'Successfully wrote {len(all_records)} audit records to {key}')
        else:
            print('No audit records found')

        # Update state
        state['last_to_ts'] = to_ts
        state['last_successful_run'] = now
        save_state(bucket, state_key, state)

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

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)}')
    return {}

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, separators=(',', ':')),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

def iso_format(ts):
    """Convert Unix timestamp to ISO 8601 format."""
    return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(ts))

def build_audit_url(api_base_url, from_ts, to_ts):
    """Build URL for TeamViewer audit API endpoint."""
    base_endpoint = f"{api_base_url.rstrip('/')}/reports/connections"
    params = {
        'from_date': iso_format(from_ts),
        'to_date': iso_format(to_ts)
    }
    query_string = urllib.parse.urlencode(params)
    return f"{base_endpoint}?{query_string}"

def fetch_audit_data(url, script_token, user_agent, http_timeout, max_retries, offset_id=None):
    """Fetch audit data from TeamViewer API with retries and pagination support."""
    # Add offset_id parameter if provided
    if offset_id:
        separator = '&' if '?' in url else '?'
        url = f"{url}{separator}offset_id={offset_id}"

    attempt = 0
    while True:
        req = urllib.request.Request(url, method='GET')
        req.add_header('User-Agent', user_agent)
        req.add_header('Authorization', f'Bearer {script_token}')
        req.add_header('Accept', 'application/json')

        try:
            with urllib.request.urlopen(req, timeout=http_timeout) as r:
                response_data = r.read()
                content_type = r.headers.get('Content-Type') or 'application/json'

                # Extract next_offset from response if present
                try:
                    data = json.loads(response_data)
                    next_offset = data.get('next_offset')
                except:
                    next_offset = None

                return response_data, content_type, next_offset

        except urllib.error.HTTPError as e:
            if e.code == 429:
                attempt += 1
                print(f'Rate limited (429) on attempt {attempt}')
                if attempt > max_retries:
                    raise
                time.sleep(min(60, 2 ** attempt) + (time.time() % 1))
            else:
                print(f'HTTP error {e.code}: {e.reason}')
                raise
        except urllib.error.URLError as e:
            attempt += 1
            print(f'URL error on attempt {attempt}: {e}')
            if attempt > max_retries:
                raise
            time.sleep(min(60, 2 ** attempt) + (time.time() % 1))

def put_audit_data(bucket, prefix, blob_data, content_type, from_ts, to_ts):
    """Write audit data to GCS."""
    ts_path = time.strftime('%Y/%m/%d', time.gmtime(to_ts))
    uniq = f"{int(time.time() * 1e6)}_{uuid.uuid4().hex[:8]}"
    key = f"{prefix}{ts_path}/teamviewer_audit_{int(from_ts)}_{int(to_ts)}_{uniq}.json"

    blob = bucket.blob(key)
    blob.metadata = {
        'source': 'teamviewer-audit',
        'from_timestamp': str(int(from_ts)),
        'to_timestamp': str(int(to_ts))
    }
    blob.upload_from_string(blob_data, content_type=content_type)

    return key

第二個檔案：requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*

點選「部署」來儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。

注意： 每次呼叫 /api/v1/reports/connections 端點時，最多只能傳回 1000 筆記錄。實作會從回應中讀取 next_offset，並使用 offset_id 進行迴圈，直到擷取所有記錄為止。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。

前往 GCP 主控台的「Cloud Scheduler」。
點選「建立工作」。

請提供下列設定詳細資料：

設定	值
名稱	`teamviewer-logs-collector-hourly`
區域	選取與 Cloud Run 函式相同的區域
頻率	`0 * * * *` (每小時整點)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	選取 Pub/Sub 主題 (`teamviewer-logs-trigger`)
郵件內文	`{}` (空白 JSON 物件)

點選「建立」。

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率：

頻率	Cron 運算式	用途
每 5 分鐘	`/5 * * *`	高容量、低延遲
每 15 分鐘檢查一次	`/15 * * *`	普通量
每小時	`0 * * * *`	標準 (建議採用)
每 6 小時	`0 /6 * *`	少量、批次處理
每日	`0 0 * * *`	歷來資料集合

測試整合項目

在 Cloud Scheduler 控制台中，找出您的作業 (teamviewer-logs-collector-hourly)。
按一下「強制執行」，手動觸發工作。
稍等幾秒鐘。
前往「Cloud Run」>「Services」。
按一下函式名稱 (teamviewer-logs-collector)。
按一下 [Logs] (記錄) 分頁標籤。

確認函式是否已順利執行。請找出以下項目：

Fetching TeamViewer audit data from YYYY-MM-DDTHH:MM:SSZ to YYYY-MM-DDTHH:MM:SSZ
Retrieved X audit records (total: X)
Successfully wrote X audit records to teamviewer/audit/YYYY/MM/DD/teamviewer_audit_...json

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (teamviewer-logs)。
前往前置字元資料夾 (teamviewer/audit/)。
確認是否已建立含有目前時間戳記的新 .json 檔案。

如果在記錄中發現錯誤：

HTTP 401：檢查 SCRIPT_TOKEN 環境變數是否與 TeamViewer 指令碼權杖相符
HTTP 403：確認指令碼權杖具有「連線報告」>「查看連線報告」權限
HTTP 429：頻率限制 - 函式會自動重試，並採用指數輪詢間隔
缺少環境變數：檢查是否已設定所有必要變數 (GCS_BUCKET、API_BASE_URL、SCRIPT_TOKEN)

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 TeamViewer logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「TeamViewer」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
複製這個電子郵件地址，以便在下一步中使用。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (teamviewer-logs)。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。

注意： 如果打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案並清空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。

在 Google SecOps 中設定動態饋給，擷取 TeamViewer 記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 TeamViewer logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「TeamViewer」做為「記錄類型」。
點選 [下一步]。
指定下列輸入參數的值：
- 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
```
gs://teamviewer-logs/teamviewer/audit/
```
  - 取代：
    - teamviewer-logs：您的 GCS bucket 名稱。
    - teamviewer/audit/：儲存記錄的前置字元/資料夾路徑。
  注意： URI 結尾一律要加上尾端斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
  - 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
  - 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    
    注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

UDM 對應表

記錄欄位	UDM 對應	邏輯
AffectedItem	metadata.product_log_id	原始記錄中的 AffectedItem 值會直接對應至這個 UDM 欄位。
EventDetails.NewValue	principal.resource.attribute.labels.value	如果 PropertyName 包含 (server)，NewValue 會做為 principal.resource.attribute.labels 中標籤的值。
EventDetails.NewValue	principal.user.user_display_name	如果 PropertyName 是參與者名稱，則 NewValue 會做為主體的顯示名稱。
EventDetails.NewValue	principal.user.userid	如果 PropertyName 是參與者的 ID，則 NewValue 會做為主體的 User-ID。
EventDetails.NewValue	security_result.about.labels.value	對於所有其他 PropertyName 值 (特定條件處理的值除外)，NewValue 會做為 security_result.about.labels 陣列中標籤的值。
EventDetails.NewValue	target.file.full_path	如果 PropertyName 是「來源檔案」，NewValue 會做為目標檔案的完整路徑。
EventDetails.NewValue	target.resource.attribute.labels.value	如果 PropertyName 包含 (client)，NewValue 會做為 target.resource.attribute.labels 中標籤的值。
EventDetails.NewValue	target.user.user_display_name	如果 PropertyName 是「Name of presenter」，系統會剖析 NewValue。如果是整數，系統會捨棄。否則，系統會將其做為目標的使用者顯示名稱。
EventDetails.NewValue	target.user.userid	如果 PropertyName 是簡報者的 ID，則 NewValue 會做為目標的使用者 ID。
EventDetails.PropertyName	principal.resource.attribute.labels.key	如果 PropertyName 包含 (server)，則 PropertyName 會做為 principal.resource.attribute.labels 中標籤的鍵。
EventDetails.PropertyName	security_result.about.labels.key	對於所有其他 PropertyName 值 (特定條件處理的值除外)，PropertyName 會做為 security_result.about.labels 陣列中標籤的鍵。
EventDetails.PropertyName	target.resource.attribute.labels.key	如果 PropertyName 包含 (client)，PropertyName 會做為 target.resource.attribute.labels 中標籤的鍵。
事件名稱	metadata.product_event_type	原始記錄中的 EventName 值會直接對應至這個 UDM 欄位。
時間戳記	metadata.event_timestamp	系統會剖析原始記錄中的時間戳記值，並做為中繼資料中的事件時間戳記。
metadata.event_type		如果 src_user (衍生自參與者 ID) 不為空白，請設為 USER_UNCATEGORIZED，否則請設為 USER_RESOURCE_ACCESS。
metadata.vendor_name		硬式編碼為 TEAMVIEWER。
metadata.product_name		硬式編碼為 TEAMVIEWER。
network.application_protocol		硬式編碼為 TEAMVIEWER。

需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。