收集 Box JSON 記錄
剖析器版本:3.0
本文說明如何使用 Google Cloud Storage,將 Box JSON 記錄檔擷取至 Google Security Operations。剖析器會處理 JSON 格式的 Box 事件記錄,並對應至統合資料模型 (UDM)。這個外掛程式會從原始記錄擷取相關欄位、執行重新命名和合併等資料轉換作業,並在輸出結構化事件資料前,以中介資訊充實資料。
事前準備
請確認您符合下列必要條件:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 具備管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- Box 的特殊權限存取權 (管理員 + 開發人員控制台)
設定 Box Developer Console (用戶端憑證)
- 登入 Box 開發人員控制台。
- 建立自訂應用程式,並啟用伺服器驗證 (用戶端憑證授權)。
- 將「Application Access」設為「App + Enterprise Access」。
- 在「Application Scopes」中,啟用「Manage enterprise properties」。
- 在「管理控制台」>「應用程式」>「自訂應用程式管理員」中,依「用戶端 ID」授權應用程式。
- 複製用戶端 ID 和用戶端密鑰,並儲存在安全的位置。
- 依序前往「管理控制台」>「帳戶與帳單」>「帳戶資訊」。
- 複製企業 ID,並儲存在安全的位置。
建立 Google Cloud Storage bucket
- 前往 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 description」(服務帳戶說明):輸入
Service account for Cloud Run function to collect Box Collaboration logs。
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions 叫用者」。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許叫用函式
授予 GCS bucket 的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (
box-collaboration-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (
- 按一下「儲存」。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會發布至該主題,Cloud Run 函式則會訂閱該主題。
- 在 GCP 控制台中,依序前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 提供下列設定詳細資料:
- 主題 ID:輸入
box-collaboration-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Box API 擷取記錄並寫入 GCS。
- 在 GCP 控制台中,前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」(使用直接修改建立函式)。
在「Configure」(設定) 部分,提供下列設定詳細資料:
設定 值 服務名稱 box-collaboration-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」或更新版本 在「Trigger (optional)」(觸發條件 (選用)) 部分:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中,選擇 Pub/Sub 主題 (
box-collaboration-trigger)。 - 按一下「儲存」。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶 (
box-collaboration-collector-sa)。
- 服務帳戶:選取服務帳戶 (
前往「容器」分頁:
- 按一下「Variables & Secrets」(變數與密鑰)。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 GCS_BUCKETbox-collaboration-logsGCS_PREFIXbox/collaboration/STATE_KEYbox/collaboration/state.jsonBOX_CLIENT_ID輸入 Box 用戶端 ID BOX_CLIENT_SECRET輸入 Box 用戶端密鑰 BOX_ENTERPRISE_ID輸入 Box 企業 ID STREAM_TYPEadmin_logs_streamingLIMIT500在「Variables & Secrets」(變數與密鑰) 部分,向下捲動至「Requests」(要求):
- 要求逾時:輸入
600秒 (10 分鐘)。
- 要求逾時:輸入
前往「設定」分頁:
- 在「Resources」(資源) 部分:
- 「記憶體」:選取「512 MiB」或以上。
- CPU:選取「1」。
- 在「Resources」(資源) 部分:
在「Revision scaling」(修訂版本資源調度) 部分:
- 「Minimum number of instances」(執行個體數量下限):輸入
0。 - 「執行個體數量上限」:輸入
100(或依預期負載調整)。
- 「Minimum number of instances」(執行個體數量下限):輸入
點選「建立」。
等待服務建立完成 (1 至 2 分鐘)。
服務建立完成後,內嵌程式碼編輯器會自動開啟。
新增函式程式碼
- 在「函式進入點」中輸入 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 分鐘)。
建立 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」>「服務」。
- 按一下函式名稱 (
box-collaboration-collector)。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。請找出以下項目:
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」>「Bucket」。
按一下 bucket 名稱。
前往前置字元資料夾 (
box/collaboration/)。確認系統已建立新的
.json檔案,並加上目前的時間戳記。
如果記錄中顯示錯誤:
- HTTP 401:檢查環境變數中的 Box API 憑證
- HTTP 403:確認 Box 應用程式具備必要權限,且已在管理控制台中獲得授權
- HTTP 429:頻率限制 - 函式會自動重試,並採用輪詢機制
- 缺少環境變數:確認已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台,請依序前往「設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中,輸入動態饋給名稱 (例如
Box Collaboration)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「Log type」中選取「Box」。
按一下「Get Service Account」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,下一個步驟將會用到。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」>「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下「儲存」。
在 Google SecOps 設定動態饋給,以便擷取 Box 記錄
- 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台,請依序前往「設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中,輸入動態饋給名稱 (例如
Box Collaboration)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「Log type」中選取「Box」。
- 點選「下一步」。
指定下列輸入參數的值:
Storage bucket URL:輸入 GCS bucket URI,並加上前置字元路徑:
gs://box-collaboration-logs/box/collaboration/更改項目:
box-collaboration-logs:GCS bucket 名稱。box/collaboration/::記錄檔的儲存前置字元/資料夾路徑。
範例:
- 根值區:
gs://company-logs/ - 前置字元:
gs://company-logs/box-logs/ - 子資料夾:
gs://company-logs/box/collaboration/
- 根值區:
來源刪除選項:根據偏好選取刪除選項:
- Never:轉移後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:在成功轉移後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後,刪除檔案和空白目錄。
檔案存在時間上限:包含過去幾天內修改過的檔案。預設值為 180 天。
Asset namespace:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「完成」畫面中檢查新的新聞動態消息設定,然後按一下「提交」。
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。 | |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.event_type |
metadata.product_event_type |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.city_name |
principal.location.city |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.country_code |
principal.location.country_or_region |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.latitude |
principal.location.region_coordinates.latitude |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.longitude |
principal.location.region_coordinates.longitude |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.region_name |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.ip |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.ip_info.registrant |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.item_id |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.item_name |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.item_path |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.item_type |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.alert_activities.0.occurred_at |
additional_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.priority |
security_result.severity |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_id |
security_result.rule_id |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.static_scan_result.family |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.static_scan_result.scan_result |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.static_scan_result.scanned_at |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.threat_info.description |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.threat_info.scanned_at |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.threat_info.source |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.threat_info.status |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.threat_info.threat_name |
security_result.detection_fields |
從變更記錄對應 |
entry.additional_details.shield_alert.user.email |
principal.user.email_addresses |
從變更記錄對應 |
entry.additional_details.shield_alert.user.id |
principal.user.userid |
從變更記錄對應 |
entry.additional_details.shield_alert.user.name |
principal.user.user_display_name |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.file_info.name |
target.file.names |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.file_info.size |
target.file.size |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.malware_info.file_info.hash |
target.file.sha1 |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.upload_activity.item_path |
target.file.full_path |
從變更記錄對應 |
entry.additional_details.shield_alert.rule_category |
security_result.category_details |
從變更記錄對應 |
entry.additional_details.shield_alert.rule_id |
security_result.rule_id |
從變更記錄對應 |
entry.additional_details.shield_alert.rule_name |
security_result.rule_name |
從變更記錄對應 |
entry.additional_details.shield_alert.risk_score |
security_result.risk_score |
從變更記錄對應 |
entry.additional_details.shield_alert.alert_summary.description |
security_result.description |
從變更記錄對應 |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專家尋求解答。