收集 Privacy-i 記錄
本文說明如何使用 Google Cloud Storage V2,將 Privacy-i 記錄檔擷取至 Google Security Operations。
Privacy-i 是由 Somansa (現已由三井物產 Secure Directions 收購) 開發的端點資料隱私權和法規遵循監控平台。這個平台可探索及分類端點上的機密資料,並強制執行政策。Privacy-i REST API 可讓您透過程式輔助方式,存取檢查記錄、事故通報和代理程式活動資料。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- Privacy-i 管理控制台的管理員權限
- Privacy-i REST API 的 API 憑證 (API 金鑰或 OAuth2 憑證)
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 privacy-i-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
收集 Privacy-i API 憑證
取得 API 金鑰
- 使用管理員憑證登入 Privacy-i 管理控制台。
- 依序前往「設定」>「API 管理」 (或「系統」>「API 設定」)。
- 點按「Generate API Key」。
- 輸入 API 金鑰名稱 (例如
Google SecOps Integration)。 複製下列詳細資料並存放在安全位置:
- API 金鑰:產生的 API 金鑰值
- 伺服器網址:Privacy-i 管理伺服器網址 (例如
https://privacy-i.your-domain.com)
判斷 API 基準網址
Privacy-i API 基準網址是管理伺服器網址:
| 格式 | 範例 |
|---|---|
| 伺服器網址 | https://privacy-i.your-domain.com/api |
測試 API 存取權
請先測試憑證,再繼續整合:
# Replace with your actual credentials PRIVACY_I_API_KEY="your-api-key" PRIVACY_I_BASE_URL="https://privacy-i.your-domain.com/api" # Test API access - list recent events curl -s -X GET "${PRIVACY_I_BASE_URL}/v1/events?limit=1" \ -H "Authorization: Bearer ${PRIVACY_I_API_KEY}" \ -H "Accept: application/json"
建立 Cloud Run 函式的服務帳戶
Cloud Run 函式需要服務帳戶,並具備寫入 GCS bucket 的權限,且可由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
privacy-i-logs-collector-sa - 服務帳戶說明:輸入
Service account for Cloud Run function to collect Privacy-i logs
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許叫用函式
授予 GCS bucket 的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Buckets」。
- 點按 bucket 名稱 (例如
privacy-i-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
privacy-i-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,供 Cloud Scheduler 發布訊息,以及 Cloud Run 函式訂閱訊息。
- 在 GCP 控制台中,前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 提供下列設定詳細資料:
- 主題 ID:輸入
privacy-i-logs-trigger - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Privacy-i REST API 擷取記錄,並將記錄寫入 GCS。
- 前往 GCP 控制台的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 privacy-i-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件 (選用)」部分:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇主題
privacy-i-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶
privacy-i-logs-collector-sa。
- 服務帳戶:選取服務帳戶
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETprivacy-i-logsGCS bucket 名稱 GCS_PREFIXprivacy-i記錄檔的前置字串 STATE_KEYprivacy-i/state.json狀態檔案路徑 PRIVACY_I_API_KEYyour-api-keyPrivacy-i API 金鑰 PRIVACY_I_BASE_URLhttps://privacy-i.your-domain.com/apiPrivacy-i API 基準網址 MAX_RECORDS5000每次執行的記錄數上限 PAGE_SIZE1000每頁記錄數 LOOKBACK_HOURS24初始回溯期 在「Variables & Secrets」(變數與密鑰) 部分,向下捲動至「Requests」(要求):
- 要求逾時:輸入
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', 'privacy-i') STATE_KEY = os.environ.get('STATE_KEY', 'privacy-i/state.json') PRIVACY_I_API_KEY = os.environ.get('PRIVACY_I_API_KEY') PRIVACY_I_BASE_URL = os.environ.get('PRIVACY_I_BASE_URL') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Privacy-i inspection and incident logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, PRIVACY_I_API_KEY, PRIVACY_I_BASE_URL]): print('Error: Missing required environment variables') 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_event_time"): try: last_time = parse_datetime(state["last_event_time"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") # Fetch logs from multiple endpoints all_records = [] newest_event_time = None for endpoint_type in ['events', 'incidents']: records, newest_time = fetch_logs( endpoint_type=endpoint_type, start_time=last_time, end_time=now, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) all_records.extend(records) if newest_time: if newest_event_time is None or parse_datetime(newest_time) > parse_datetime(newest_event_time): newest_event_time = newest_time if not all_records: print("No new 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 with newest event time if newest_event_time: save_state(bucket, STATE_KEY, newest_event_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(all_records)} records") except Exception as e: print(f'Error processing 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: {e}") return {} def save_state(bucket, key, last_event_time_iso: str): """Save the last event timestamp to GCS state file.""" try: state = {'last_event_time': last_event_time_iso} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_event_time={last_event_time_iso}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(endpoint_type: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int): """ Fetch logs from Privacy-i REST API with offset-based pagination and rate limiting. Args: endpoint_type: API endpoint type (events, incidents) start_time: Start time for log query end_time: End time for log query page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ api_base = PRIVACY_I_BASE_URL.rstrip('/') endpoint = f"{api_base}/v1/{endpoint_type}" headers = { 'Authorization': f'Bearer {PRIVACY_I_API_KEY}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-PrivacyICollector/1.0' } records = [] newest_time = None page_num = 0 offset = 0 backoff = 1.0 start_iso = start_time.strftime('%Y-%m-%dT%H:%M:%SZ') end_iso = end_time.strftime('%Y-%m-%dT%H:%M:%SZ') while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records}) for {endpoint_type}") break url = f"{endpoint}?startDate={start_iso}&endDate={end_iso}&limit={min(page_size, max_records - len(records))}&offset={offset}" try: response = http.request('GET', url, headers=headers) # Handle rate limiting with exponential backoff 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, 30.0) continue backoff = 1.0 if response.status != 200: print(f"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return records, newest_time data = json.loads(response.data.decode('utf-8')) page_results = data.get('data', data.get('results', data.get('items', []))) if not page_results: print(f"No more results (empty page) for {endpoint_type}") break print(f"{endpoint_type} page {page_num}: Retrieved {len(page_results)} events") # Add endpoint type for identification for event in page_results: event['_privacy_i_log_type'] = endpoint_type records.extend(page_results) # Track newest event time for event in page_results: try: event_ts = event.get('timestamp') or event.get('eventTime') or event.get('createdAt') if event_ts: event_time = str(event_ts) if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for more results total = data.get('total', data.get('totalCount', 0)) if total and (offset + len(page_results)) >= total: print(f"No more pages for {endpoint_type} (reached total)") break if len(page_results) < page_size: print(f"No more pages for {endpoint_type} (last page not full)") break offset += len(page_results) except Exception as e: print(f"Error fetching {endpoint_type} logs: {e}") return records, newest_time print(f"Retrieved {len(records)} total {endpoint_type} records from {page_num} pages") return records, newest_time第二個檔案 - 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」。
- 點選「建立工作」。
提供下列設定詳細資料:
設定 值 名稱 privacy-i-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題「 privacy-i-logs-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
請根據記錄量和延遲時間需求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
高流量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等音量 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
低量、批次處理 |
| 每天 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中找到工作。
- 按一下「強制執行」,手動觸發工作。
- 稍等幾秒鐘。
- 依序前往「Cloud Run」 > 「服務」。
- 按一下
privacy-i-logs-collector。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。尋找:
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 events page 1: Retrieved X events incidents page 1: Retrieved X events Wrote X records to gs://privacy-i-logs/privacy-i/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
點選 bucket 名稱 (
privacy-i-logs)。前往
privacy-i/資料夾。確認系統已建立新的
.ndjson檔案,並加上目前的時間戳記。
如果記錄中顯示錯誤:
- HTTP 401:檢查環境變數中的 API 金鑰
- HTTP 403:確認 API 金鑰在 Privacy-i 管理控制台中具備必要權限
- HTTP 429:頻率限制 - 函式會自動重試,並採用輪詢機制
- 缺少環境變數:確認已設定所有必要變數
在 Google SecOps 中設定動態消息,擷取 Privacy-i 記錄
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中,輸入動態饋給名稱 (例如
Privacy-i Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Privacy-i」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址。
點選「下一步」。
指定下列輸入參數的值:
Storage bucket URL:輸入 GCS bucket URI,並加上前置字元路徑:
gs://privacy-i-logs/privacy-i/- 取代:
privacy-i-logs:您的 GCS bucket 名稱。privacy-i:儲存記錄的選用前置字元/資料夾路徑 (如為根目錄,請留空)。
- 取代:
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:一律不會在轉移後刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移檔案後,刪除檔案和空白目錄。
檔案存在時間上限:納入在過去天數內修改的檔案 (預設為 180 天)
資產命名空間:資產命名空間
擷取標籤:要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
- Google SecOps 服務帳戶必須具備 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 點按 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色:選取「Storage 物件檢視者」
- 按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| has_principal、has_target_file、has_user | metadata.event_type | 如果 has_principal 和 has_target_file,則設為 SCAN_FILE;如果 has_user,則設為 USER_UNCATEGORIZED;如果 has_principal,則設為 STATUS_UPDATE;否則設為 GENERIC_EVENT |
| elog_incidentlogguid、efile_incidentlogguid、guid | metadata.product_log_id | elog_incidentlogguid 的值 (如果不是空白),否則為 efile_incidentlogguid 的值 (如果不是空白),否則為 guid |
| 資料類型 | metadata.product_event_type | 直接複製值 |
| productversion | metadata.product_version | 直接複製值 |
| metadata.product_name | 設為「PRIVACY_I」 | |
| metadata.vendor_name | 設為「PRIVACY_I」 | |
| dfile_agentip, ip | principal.asset.ip | 如果 dfile_agentip 不為空白,則從 dfile_agentip 合併,否則從 ip 合併 |
| dfile_agentip、ip | principal.ip | 如果 dfile_agentip 不為空白,則從 dfile_agentip 合併,否則從 ip 合併 |
| dfile_computername | principal.asset.asset_id | 串連為「Computer:%{dfile_computername}」 |
| 電腦名稱 | principal.asset.hostname | 串連為「Computer:%{computername}」 |
| osserial_label, computerguid_label | principal.asset.attribute.labels | 從 osserial_label 和 computerguid_label 合併 |
| productid | principal.asset.product_object_id | 直接複製值 |
| dlog_ostype、osname | principal.platform | 根據 dlog_ostype 列舉 (0=WINDOWS、1=LINUX、2=MAC) 設定,然後根據 osname 規則運算式覆寫 |
| macaddr | principal.mac | 從 macaddr 合併 |
| dfile_userid、userid、admin_id | principal.user.userid | 設為 dfile_userid,然後在不為空值時由 userid 覆寫,接著在不為空值時由 admin_id 覆寫 |
| dfile_username、empname、admin_name | principal.user.user_display_name | 設為 dfile_username,然後由 empname (如不為空白) 覆寫,再由 admin_name (如不為空白) 覆寫 |
| dfile_deptname | principal.user.department | 直接複製值 |
| logintime | principal.user.last_login_time | 使用日期格式 yyyy-MM-dd HH:mm:ss 轉換 |
| 工作 | principal.user.title | 直接複製值 |
| dfile_patterninfo | security_result | 從 JSON 擷取並整理為偵測欄位 |
| category | security_result.category_details | 直接複製值 |
| logtype | security_result.action | 直接複製值 |
| elog_detectedapplicationname | target.application | 直接複製值 |
| dfile_filecreatedtime | target.file.create_time | 使用 ISO8601 日期格式轉換 |
| dfile_fileextension | target.file.mime_type | 直接複製值 |
| dfile_filename | target.file.names | 從 dfile_filename 合併 |
| dfile_filesize | target.file.size | 已轉換為整數 |
| dfile_firstscannedpath | target.file.full_path | 直接複製值 |
| dfile_firstscannedtime | target.file.first_seen_time | 使用 ISO8601 日期格式轉換 |
| dfile_filemodifiedtime | target.file.last_modification_time | 使用 ISO8601 日期格式轉換 |
| efile_fileaccessedtime | target.file.last_access_time | 使用日期格式 yyyy-MM-dd HH:mm:ss 轉換 |
| efile_resulttime | target.file.last_analysis_time | 使用 ISO8601 日期格式轉換 |
| efile_patternguid_label、dfile_dataformatname_label、dfile_filerelationguid_label、dfile_inspectcount_label、dfile_inspectendtime_label、dfile_inspectstarttime_label、dfile_inspectfilecount_label、dfile_inspectpropertycount_label、dfile_inspectpropertyfilecount_label、dfile_inspectlogtime_label、dfile_filetype_label、dlog_inspectinfo_label、dlog_inspectexaminationtype_label、dlog_inspecttype_label、dfile_encryptfiletype_label、dfile_scanlogtype_label、dfile_policyname_label、elog_mediaguid_label、elog_mediaid_label、loggroup_label、agent_id_label、logouttime_label、packageinstalledtime_label、registeredtime_label、detail_label | additional.fields | 從多個標籤欄位合併 |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求解答。