收集 Box JSON 記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，將 Box JSON 記錄檔擷取至 Google Security Operations。剖析器會處理 JSON 格式的 Box 事件記錄，並將其對應至統一資料模型 (UDM)。這項功能會從原始記錄中擷取相關欄位，執行重新命名和合併等資料轉換作業，並在輸出結構化事件資料前，以中介資訊擴充資料。

事前準備

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

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
Box 的特殊存取權 (管理員和開發人員控制台)

設定 Box 開發人員控制台 (用戶端憑證)

登入 Box 開發人員控制台。
使用伺服器驗證 (用戶端憑證授權) 建立自訂應用程式。
將「Application Access」設為「App + Enterprise Access」。
在「應用程式範圍」中，啟用「管理企業資源」。
在管理控制台 >「應用程式」>「自訂應用程式管理工具」中，使用用戶端 ID 授權應用程式。
複製「Client ID」(用戶端 ID) 和「Client Secret」(用戶端密鑰)，並儲存在安全地點。
依序前往「管理控制台」>「帳戶和帳單」>「帳戶資訊」。
複製並儲存企業 ID，並妥善保管。

建立 Google Cloud Storage 值區

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

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

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

點選「建立」。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於：

Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
Cloud Run 叫用者：允許 Pub/Sub 叫用函式
Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

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

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

建立 Pub/Sub 主題

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

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

建立 Cloud Run 函式來收集記錄

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

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

設定值

服務名稱 box-collaboration-collector

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

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

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

前往「容器」分頁：

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

變數名稱	範例值
`GCS_BUCKET`	`box-collaboration-logs`
`GCS_PREFIX`	`box/collaboration/`
`STATE_KEY`	`box/collaboration/state.json`
`BOX_CLIENT_ID`	輸入 Box 用戶端 ID
`BOX_CLIENT_SECRET`	輸入 Box 用戶端密鑰
`BOX_ENTERPRISE_ID`	輸入 Box 企業 ID
`STREAM_TYPE`	`admin_logs_streaming`
`LIMIT`	`500`

在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入 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, timezone
import base64

# 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()

TOKEN_URL = "https://api.box.com/oauth2/token"
EVENTS_URL = "https://api.box.com/2.0/events"

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Box enterprise events 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', 'box/collaboration/')
    state_key = os.environ.get('STATE_KEY', 'box/collaboration/state.json')

    client_id = os.environ.get('BOX_CLIENT_ID')
    client_secret = os.environ.get('BOX_CLIENT_SECRET')
    enterprise_id = os.environ.get('BOX_ENTERPRISE_ID')
    stream_type = os.environ.get('STREAM_TYPE', 'admin_logs_streaming')
    limit = int(os.environ.get('LIMIT', '500'))

    if not all([bucket_name, client_id, client_secret, enterprise_id]):
        print('Error: Missing required environment variables')
        return

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

        # Get OAuth token
        token = get_token(client_id, client_secret, enterprise_id)

        # Load state (stream position)
        state = load_state(bucket, state_key)
        stream_position = state.get('stream_position')

        print(f'Processing events from stream position: {stream_position}')

        total_events = 0
        idx = 0

        while True:
            # Fetch events page
            page = fetch_events(token, stream_type, limit, stream_position)
            entries = page.get('entries') or []

            if not entries:
                next_pos = page.get('next_stream_position') or stream_position
                if next_pos and next_pos != stream_position:
                    save_state(bucket, state_key, {'stream_position': next_pos})
                break

            # Write page to GCS
            timestamp = datetime.now(timezone.utc).strftime('%Y/%m/%d/%H%M%S')
            blob_name = f"{prefix}{timestamp}-box-events-{idx:03d}.json"
            blob = bucket.blob(blob_name)
            blob.upload_from_string(
                json.dumps(page, separators=(',', ':')),
                content_type='application/json'
            )

            idx += 1
            total_events += len(entries)
            stream_position = page.get('next_stream_position') or stream_position

            # Save state after each page
            if stream_position:
                save_state(bucket, state_key, {'stream_position': stream_position})

            # Break if fewer entries than limit (last page)
            if len(entries) < limit:
                break

        print(f'Successfully processed {total_events} events, final position: {stream_position}')

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

def get_token(client_id, client_secret, enterprise_id):
    """Get OAuth 2.0 access token using client credentials grant."""
    fields = {
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
        'box_subject_type': 'enterprise',
        'box_subject_id': enterprise_id
    }

    response = http.request(
        'POST',
        TOKEN_URL,
        fields=fields,
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )

    token_data = json.loads(response.data.decode('utf-8'))
    return token_data['access_token']

def fetch_events(token, stream_type, limit, stream_position=None, timeout=60, max_retries=5):
    """Fetch events from Box API with retry logic."""
    params = {
        'stream_type': stream_type,
        'limit': str(limit),
        'stream_position': stream_position or 'now'
    }

    # Build query string
    query_string = '&'.join([f"{k}={v}" for k, v in params.items()])
    url = f"{EVENTS_URL}?{query_string}"

    attempt = 0
    backoff = 1.0

    while True:
        try:
            response = http.request(
                'GET',
                url,
                headers={'Authorization': f'Bearer {token}'},
                timeout=timeout
            )

            if response.status == 200:
                return json.loads(response.data.decode('utf-8'))
            elif response.status == 429 and attempt < max_retries:
                # Rate limited - retry with backoff
                retry_after = response.headers.get('Retry-After')
                delay = int(retry_after) if retry_after and retry_after.isdigit() else int(backoff)
                print(f'Rate limited, retrying after {delay} seconds')
                import time
                time.sleep(max(1, delay))
                attempt += 1
                backoff *= 2
                continue
            elif 500 <= response.status <= 599 and attempt < max_retries:
                # Server error - retry with backoff
                print(f'Server error {response.status}, retrying after {backoff} seconds')
                import time
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            else:
                raise Exception(f'Box API error: {response.status} {response.data.decode("utf-8")}')
        except Exception as e:
            if attempt < max_retries:
                print(f'Request error: {str(e)}, retrying after {backoff} seconds')
                import time
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            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)}')

第二個檔案：requirements.txt:

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

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

注意： Pub/Sub 觸發條件設定會自動建立必要的訂閱項目和權限。

建立 Cloud Scheduler 工作

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

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

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

設定	值
名稱	`box-collaboration-schedule-15min`
區域	選取與 Cloud Run 函式相同的區域
頻率	`/15 * * *` (每 15 分鐘)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	選取 Pub/Sub 主題 (`box-collaboration-trigger`)
郵件內文	`{}` (空白 JSON 物件)

點選「建立」。

排程頻率選項

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

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

測試整合項目

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

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

Processing events from stream position: ...
Page 1: Retrieved X events
Wrote X records to gs://box-collaboration-logs/box/collaboration/...
Successfully processed X events

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

如果在記錄中發現錯誤：

HTTP 401：檢查環境變數中的 Box API 憑證
HTTP 403：確認 Box 應用程式具備必要權限，且已在管理控制台中獲得授權
HTTP 429：頻率限制 - 函式會自動重試並延遲
缺少環境變數：檢查是否已設定所有必要變數

擷取 Google SecOps 服務帳戶

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

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

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

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

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

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

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

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

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

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

UDM 對應表

記錄欄位	UDM 對應	邏輯
additional_details.ekm_id	additional.fields	取自 additional_details.ekm_id 的值
additional_details.service_id	additional.fields	取自 additional_details.service_id 的值
additional_details.service_name	additional.fields	取自 additional_details.service_name 的值
additional_details.shared_link_id	additional.fields	取自 additional_details.shared_link_id 的值
additional_details.size	target.file.size	取自 additional_details.size 的值
additional_details.version_id	additional.fields	取自 additional_details.version_id 的值
created_at	metadata.event_timestamp	取自 created_at 的值
created_by.id	principal.user.userid	取自 created_by.id 的值
created_by.login	principal.user.email_addresses	取自 created_by.login 的值
created_by.name	principal.user.user_display_name	取自 created_by.name 的值
event_id	metadata.product_log_id	從 event_id 取得的值
event_type	metadata.product_event_type	從 event_type 取得的值
ip_address	principal.ip	取自 ip_address 的值
source.item_id	target.file.product_object_id	取自 source.item_id 的值
source.item_name	target.file.full_path	取自 source.item_name 的值
source.item_type	未對應
source.login	target.user.email_addresses	取自 source.login 的值
source.name	target.user.user_display_name	取自 source.name 的值
source.owned_by.id	target.user.userid	取自 source.owned_by.id 的值
source.owned_by.login	target.user.email_addresses	取自 source.owned_by.login 的值
source.owned_by.name	target.user.user_display_name	取自 source.owned_by.name 的值
source.parent.id	未對應
source.parent.name	未對應
source.parent.type	未對應
source.type	未對應
類型	metadata.log_type	從類型取得的值
metadata.vendor_name		硬式編碼值
metadata.product_name		硬式編碼值
security_result.action		衍生自 event_type。如果 event_type 為 FAILED_LOGIN，則為 BLOCK；如果 event_type 為 USER_LOGIN，則為 ALLOW；否則為 UNSPECIFIED。
extensions.auth.type		衍生自 event_type。如果 event_type 為 USER_LOGIN 或 ADMIN_LOGIN，則為 MACHINE，否則為 UNSPECIFIED。
extensions.auth.mechanism		衍生自 event_type。如果 event_type 為 USER_LOGIN 或 ADMIN_LOGIN，則為 USERNAME_PASSWORD，否則為 UNSPECIFIED。

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