收集 Asset Panda 記錄
剖析器版本:1.0
本文說明如何使用 Google Cloud Storage V2,將 Asset Panda 記錄擷取至 Google Security Operations。
Asset Panda 是雲端資產管理平台,可協助機構在實體和數位資產的整個生命週期中,追蹤、管理及回報這些資產。Asset Panda REST API 可存取變更記錄,記錄所有資產群組的欄位層級修改、資產轉移、動作記錄和使用者活動。您可以透過 API 收集這些變更記錄,並寫入 GCS bucket,供 Google SecOps 擷取。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 具備管理員存取權的 Asset Panda 帳戶 (API 設定時必須具備這項權限)
- 具有讀取權限的 Asset Panda API 金鑰和密鑰
產生 Asset Panda API 憑證
前往 API 設定
- 以管理員身分登入 Asset Panda 帳戶。
- 按一下右上角的「設定」 (齒輪) 圖示。
- 選取「API 設定」。
建立 API 金鑰
- 在「API 設定」頁面中,按一下「建立新的 API 金鑰」。
請提供下列設定詳細資料:
- 名稱:輸入金鑰的描述性名稱 (例如
SIEM Change Log Collector) - 權限:選取「讀取」
- 名稱:輸入金鑰的描述性名稱 (例如
按一下 [儲存]。
記錄頁面底部顯示的下列憑證:
- API 金鑰 (也稱為用戶端 ID)
- API 密鑰 (也稱為用戶端密鑰)
找出群組 ID
Cloud Run 函式需要群組 ID,才能查詢物件及其變更記錄。如要找出群組 ID,請按照下列步驟操作:
- 在 Asset Panda 中,依序前往「Settings」>「Group Settings」。
- 按一下群組名稱旁邊的「編輯」連結。
- 請注意瀏覽器網址結尾附近的數值。這是群組 ID (例如
12345)。 - 針對要收集變更記錄的每個群組重複執行上述步驟。
測試 API 存取權
請先測試憑證,再繼續進行整合:
API_KEY="your-api-key" API_SECRET="your-api-secret" # Test authentication by retrieving account settings curl -s "https://api.assetpanda.com/v3/settings" \ -H "Accept: application/json" \ -H "Access-Key-Id: ${API_KEY}" \ -H "Access-Key-Secret: ${API_SECRET}"
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 asset-panda-change-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限,且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
asset-panda-logs-collector-sa - 服務帳戶說明:輸入
Service account for Cloud Run function to collect Asset Panda change logs
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS bucket,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
asset-panda-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
asset-panda-logs-trigger - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 傳送的 Pub/Sub 訊息觸發,從 Asset Panda REST API 擷取變更記錄,並將記錄寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 asset-panda-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇主題
asset-panda-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器」、「網路」和「安全性」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶
asset-panda-logs-collector-sa
- 服務帳戶:選取服務帳戶
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETasset-panda-change-logsGCS bucket 名稱 GCS_PREFIXasset-panda記錄檔的前置字串 STATE_KEYasset-panda/state.json狀態檔案路徑 AP_API_KEYyour-api-keyAsset Panda API 金鑰 AP_API_SECRETyour-api-secretAsset Panda API 密鑰 AP_GROUP_IDS12345,67890要監控的群組 ID,以半形逗號分隔 PAGE_SIZE50每個 API 頁面的記錄數 (最多 50 筆) MAX_RECORDS10000每次執行的記錄數上限 LOOKBACK_HOURS2初始回溯期 在「變數與密鑰」部分中,向下捲動至「要求」:
- 要求逾時:輸入
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, timezone, timedelta import time # 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', 'asset-panda').strip('/') STATE_KEY = os.environ.get('STATE_KEY') or f"{GCS_PREFIX}/state.json" AP_API_KEY = os.environ.get('AP_API_KEY') AP_API_SECRET = os.environ.get('AP_API_SECRET') AP_GROUP_IDS = os.environ.get('AP_GROUP_IDS', '') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '50')) MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '2')) BASE_URL = 'https://api.assetpanda.com' # Rate limiting: 400 calls per 3 minutes RATE_LIMIT_CALLS = 400 RATE_LIMIT_WINDOW = 180 # seconds call_timestamps = [] def rate_limit_wait(): """Enforce rate limiting of 400 calls per 3-minute window.""" global call_timestamps now = time.time() call_timestamps = [t for t in call_timestamps if now - t < RATE_LIMIT_WINDOW] if len(call_timestamps) >= RATE_LIMIT_CALLS: sleep_time = RATE_LIMIT_WINDOW - (now - call_timestamps[0]) + 1 print(f"Rate limit approaching. Sleeping {sleep_time:.1f}s...") time.sleep(sleep_time) call_timestamps = [t for t in call_timestamps if time.time() - t < RATE_LIMIT_WINDOW] call_timestamps.append(time.time()) def api_request(method, path, params=None): """Make an authenticated request to the Asset Panda API.""" rate_limit_wait() url = f"{BASE_URL}{path}" headers = { 'Accept': 'application/json', 'Content-Type': 'application/json', 'Access-Key-Id': AP_API_KEY, 'Access-Key-Secret': AP_API_SECRET, } backoff = 1.0 for attempt in range(3): try: if method == 'GET': response = http.request(method, url, headers=headers, fields=params) else: body = json.dumps(params or {}).encode('utf-8') response = http.request(method, url, headers=headers, body=body) if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 60.0) continue if response.status == 200: return json.loads(response.data.decode('utf-8')) print(f"HTTP {response.status}: {response.data.decode('utf-8')[:500]}") return None except Exception as e: print(f"Request error (attempt {attempt + 1}): {e}") if attempt < 2: time.sleep(backoff) backoff *= 2 return None def parse_datetime(value): """Parse ISO datetime string to datetime object.""" if not value: return None if value.endswith("Z"): value = value[:-1] + "+00:00" try: return datetime.fromisoformat(value) except Exception: return None @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Asset Panda change logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, AP_API_KEY, AP_API_SECRET, AP_GROUP_IDS]): print('Error: Missing required environment variables') return group_ids = [gid.strip() for gid in AP_GROUP_IDS.split(',') if gid.strip()] if not group_ids: print('Error: No group IDs configured') return try: bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) last_time = None if isinstance(state, dict) and state.get("last_run_time"): try: last_time = parse_datetime(state["last_run_time"]) if last_time: last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_run_time: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Collecting change logs from {last_time.isoformat()} to {now.isoformat()}") all_records = [] for group_id in group_ids: if len(all_records) >= MAX_RECORDS: print(f"Reached max_records limit ({MAX_RECORDS})") break records = fetch_group_change_logs(group_id, last_time, now) all_records.extend(records) print(f"Group {group_id}: collected {len(records)} change log entries") if not all_records: print("No new change log records found.") save_state(bucket, STATE_KEY, now.isoformat()) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(record, ensure_ascii=False) for record in all_records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(all_records)} records") except Exception as e: print(f'Error processing change logs: {str(e)}') raise def fetch_group_change_logs(group_id, start_time, end_time): """ Fetch change logs for all objects in a group. Args: group_id: Asset Panda group ID start_time: Start time for log query end_time: End time for log query Returns: List of change log records """ records = [] offset = 0 # Iterate through objects in the group while True: if len(records) >= MAX_RECORDS: break search_data = api_request( 'POST', f'/v3/groups/{group_id}/search_objects', {'offset': offset, 'limit': PAGE_SIZE} ) if not search_data: break objects = search_data.get('objects', []) if not objects: break for obj in objects: object_id = obj.get('id') if not object_id: continue change_logs = fetch_object_change_logs( object_id, group_id, start_time, end_time ) records.extend(change_logs) if len(records) >= MAX_RECORDS: break total = search_data.get('totals', {}).get('objects', 0) offset += len(objects) if offset >= total: break return records def fetch_object_change_logs(object_id, group_id, start_time, end_time): """ Fetch change logs for a specific object. Args: object_id: Asset Panda object ID group_id: Asset Panda group ID start_time: Start time filter end_time: End time filter Returns: List of change log records with metadata """ records = [] offset = 0 while True: data = api_request( 'GET', f'/v3/entity_objects/{object_id}/change_logs', {'limit': PAGE_SIZE, 'offset': offset} ) if not data: break logs = data.get('change_logs', data.get('data', [])) if isinstance(logs, dict): logs = [logs] if not logs: break for log_entry in logs: log_time = parse_datetime( log_entry.get('created_at') or log_entry.get('updated_at', '') ) if log_time and log_time < start_time: return records if log_time and log_time > end_time: continue log_entry['_group_id'] = group_id log_entry['_object_id'] = object_id records.append(log_entry) if len(logs) < PAGE_SIZE: break offset += len(logs) return records 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: {e}") return {} def save_state(bucket, key, last_run_time_iso): """Save the last run timestamp to GCS state file.""" try: state = {'last_run_time': last_run_time_iso} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_run_time={last_run_time_iso}") except Exception as e: print(f"Warning: Could not save state: {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」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 asset-panda-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 asset-panda-logs-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 2 小時 | 0 */2 * * * |
調低音量 |
| 每 6 小時 | 0 */6 * * * |
少量、批次處理 |
測試整合項目
- 在 Cloud Scheduler 控制台中找出您的工作。
- 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱
asset-panda-logs-collector。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找:
Collecting change logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Group 12345: collected X change log entries Wrote X records to gs://asset-panda-change-logs/asset-panda/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往前置字元資料夾
asset-panda/。確認是否已建立新的
.ndjson檔案,且檔案名稱包含目前的時間戳記。
如果記錄中顯示錯誤:
- HTTP 401:檢查環境變數中的 API 憑證。確認 API 金鑰和密鑰正確無誤。
- HTTP 403:確認 API 金鑰已啟用讀取權限。
- HTTP 429:頻率限制 - 函式會自動重試,並延遲一段時間。
- 結果為空:確認已設定的群組 ID 存在,且包含具有變更記錄的物件。
- 缺少環境變數:請檢查是否已設定所有必要變數。
在 Google SecOps 中設定動態饋給,擷取 Asset Panda 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Asset Panda Change Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Asset Panda」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://asset-panda-change-logs/asset-panda/- 取代:
asset-panda-change-logs:您的 GCS bucket 名稱。asset-panda:選用的前置字元/資料夾路徑,用於儲存記錄 (如為根目錄,請留空)。
- 取代:
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
- 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)
資產命名空間:資產命名空間
擷取標籤:要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色:選取「Storage 物件檢視者」
按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
array_values_label |
additional.fields |
已合併 |
data_label |
additional.fields |
已合併 |
field_values_label |
additional.fields |
已合併 |
low_values_label |
additional.fields |
已合併 |
value_ids_label |
additional.fields |
已合併 |
values_label |
additional.fields |
已合併 |
display_with_secondary |
intermediary.user.user_display_name |
直接對應 |
secondary_name |
intermediary.user.userid |
直接對應 |
created_at |
metadata.event_timestamp |
已剖析為 RFC 3339 |
has_user |
metadata.event_type |
已對應:true → USER_UNCATEGORIZED |
object_version_ids |
metadata.product_version |
直接對應 |
share_url |
metadata.url_back_to_product |
直接對應 |
display_name |
principal.user.user_display_name |
直接對應 |
id |
principal.user.userid |
直接對應 |
account_id_label |
security_result.detection_fields |
已合併 |
change_source_label |
security_result.detection_fields |
已合併 |
change_trigger_label |
security_result.detection_fields |
已合併 |
created_at_label |
security_result.detection_fields |
已合併 |
date_format_label |
security_result.detection_fields |
已合併 |
docusign_envelope_id_label |
security_result.detection_fields |
已合併 |
embedded_into_object_id_label |
security_result.detection_fields |
已合併 |
entity_action_id_label |
security_result.detection_fields |
已合併 |
entity_id_label |
security_result.detection_fields |
已合併 |
entity_id_label1 |
security_result.detection_fields |
已合併 |
entity_key_label |
security_result.detection_fields |
已合併 |
google_calendar_sync_label |
security_result.detection_fields |
已合併 |
gps_coordinates_label |
security_result.detection_fields |
已合併 |
has_audit_history_label |
security_result.detection_fields |
已合併 |
id_label |
security_result.detection_fields |
已合併 |
is_archived_label |
security_result.detection_fields |
已合併 |
is_deletable_label |
security_result.detection_fields |
已合併 |
is_editable_label |
security_result.detection_fields |
已合併 |
is_locked_label |
security_result.detection_fields |
已合併 |
linked_action_object_id_label |
security_result.detection_fields |
已合併 |
modifier_id_label |
security_result.detection_fields |
已合併 |
next_step_reservation_uid_label |
security_result.detection_fields |
已合併 |
object_appreciation_label |
security_result.detection_fields |
已合併 |
object_depreciation_label |
security_result.detection_fields |
已合併 |
oid_label |
security_result.detection_fields |
已合併 |
old_id_label |
security_result.detection_fields |
已合併 |
parent_action_object_id_label |
security_result.detection_fields |
已合併 |
predefined_forms_label |
security_result.detection_fields |
已合併 |
reservation_notification_label |
security_result.detection_fields |
已合併 |
reservation_uid_label |
security_result.detection_fields |
已合併 |
returned_label |
security_result.detection_fields |
已合併 |
state_label |
security_result.detection_fields |
已合併 |
status_label |
security_result.detection_fields |
已合併 |
updated_at_label |
security_result.detection_fields |
已合併 |
user_id_label |
security_result.detection_fields |
已合併 |
version_label |
security_result.detection_fields |
已合併 |
| 不適用 | metadata.event_type |
常數:USER_UNCATEGORIZED |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。