收集 FortiCNAPP (原稱 Lacework) 記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage V2，將 FortiCNAPP (舊稱 Lacework) 記錄檔擷取至 Google Security Operations。
FortiCNAPP 是雲端原生應用程式保護平台 (CNAPP)，可在多雲端環境中提供雲端安全防護機制管理、工作負載保護和威脅偵測功能。並產生可透過 Lacework REST API 收集的快訊、法規遵循結果和稽核記錄。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 具備管理員權限的 FortiCNAPP (前身為 Lacework) 控制台特殊存取權
- 已啟用 API 金鑰存取權的 Lacework 帳戶
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。
請提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 lacework-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或資料保留政策 點選「建立」。
收集 FortiCNAPP (前身為 Lacework) API 憑證
產生 API 金鑰
- 登入 Lacework 控制台。
- 依序前往「設定」>「設定」>「API 金鑰」。
- 按一下「+ 新增」。
- 輸入 API 金鑰名稱 (例如
Google SecOps Integration)。 - 視需要輸入說明。
按一下 [儲存]。
複製下列詳細資料並存放在安全的位置：
- 金鑰 ID：產生的 API 金鑰 ID
- 密鑰：系統產生的 API 密鑰 (只會顯示一次)
記下瀏覽器網址列中的 Lacework 帳戶網址。
- 格式：
https://<ACCOUNT>.lacework.net - 範例：如果 Lacework 控制台網址為
https://acme.lacework.net，則帳戶名稱為acme
- 格式：
驗證權限
如要確認帳戶是否具備必要權限，請按照下列步驟操作：
- 登入 Lacework 控制台。
- 依序前往「設定」>「設定」>「API 金鑰」。
- 如果可以查看「API 金鑰」頁面並建立金鑰，表示您具備必要權限。
- 如果沒有看到這個選項，請與管理員聯絡，要求對方授予管理員層級的存取權。
測試 API 存取權
請先測試憑證，再繼續進行整合：
# Replace with your actual credentials LW_ACCOUNT="your-account-name" LW_KEY_ID="your-api-key-id" LW_SECRET="your-api-secret" # Get a temporary access token TOKEN=$(curl -s -X POST "https://${LW_ACCOUNT}.lacework.net/api/v2/access/tokens" \ -H "X-LW-UAKS: ${LW_SECRET}" \ -H "Content-Type: application/json" \ -d "{\"keyId\": \"${LW_KEY_ID}\", \"expiryTime\": 3600}" | python3 -c "import sys,json; print(json.load(sys.stdin).get('token',''))") # Test API access - list alerts curl -v -H "Authorization: Bearer ${TOKEN}" \ "https://${LW_ACCOUNT}.lacework.net/api/v2/Alerts?startTime=$(date -u -v-1d +%Y-%m-%dT%H:%M:%SZ)&endTime=$(date -u +%Y-%m-%dT%H:%M:%SZ)"
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限，且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下「Create Service Account」(建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
lacework-logs-collector-sa - 服務帳戶說明：輸入
Service account for Cloud Run function to collect FortiCNAPP (formerly Lacework) 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 名稱 (例如
lacework-logs)。 - 前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
lacework-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色：選取「Storage 物件管理員」
- 新增主體：輸入服務帳戶電子郵件地址 (例如
按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入
lacework-logs-trigger - 其他設定保留預設值
- 主題 ID：輸入
點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 FortiCNAPP (舊稱 Lacework) API 擷取記錄，並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 lacework-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇主題
lacework-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器」、「網路」和「安全性」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶
lacework-logs-collector-sa。
- 服務帳戶：選取服務帳戶
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETlacework-logsGCS bucket 名稱 GCS_PREFIXlacework記錄檔的前置字串 STATE_KEYlacework/state.json狀態檔案路徑 LW_ACCOUNTacmeLacework 帳戶名稱 LW_KEY_IDyour-api-key-idLacework API 金鑰 ID LW_SECRETyour-api-secretLacework API 密鑰 MAX_RECORDS5000每次執行的記錄數上限 PAGE_SIZE500每頁記錄數 LOOKBACK_HOURS24初始回溯期 在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入
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', 'lacework') STATE_KEY = os.environ.get('STATE_KEY', 'lacework/state.json') LW_ACCOUNT = os.environ.get('LW_ACCOUNT') LW_KEY_ID = os.environ.get('LW_KEY_ID') LW_SECRET = os.environ.get('LW_SECRET') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '500')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) # Lacework API base URL API_BASE_TEMPLATE = 'https://{account}.lacework.net/api/v2' # Log endpoints to fetch ENDPOINTS = [ {'name': 'alerts', 'path': '/Alerts', 'time_field': 'startTime', 'results_key': 'data'}, {'name': 'audit_logs', 'path': '/AuditLogs', 'time_field': 'createdTime', 'results_key': 'data'}, ] def get_access_token(api_base: str, key_id: str, secret: str) -> str: """Get a temporary access token from Lacework API.""" token_url = f"{api_base}/access/tokens" body = json.dumps({ 'keyId': key_id, 'expiryTime': 3600 }).encode('utf-8') headers = { 'X-LW-UAKS': secret, 'Content-Type': 'application/json', } response = http.request('POST', token_url, body=body, headers=headers) if response.status != 201: raise Exception(f"Failed to get access token: HTTP {response.status} - {response.data.decode('utf-8')}") token_data = json.loads(response.data.decode('utf-8')) return token_data['token'] @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch FortiCNAPP (formerly Lacework) logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, LW_ACCOUNT, LW_KEY_ID, LW_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) api_base = API_BASE_TEMPLATE.format(account=LW_ACCOUNT) # Get access token token = get_access_token(api_base, LW_KEY_ID, LW_SECRET) print("Successfully obtained access token") # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) all_records = [] for endpoint in ENDPOINTS: ep_name = endpoint['name'] last_time_str = None if isinstance(state, dict) and state.get(f"last_{ep_name}_time"): try: last_time = parse_datetime(state[f"last_{ep_name}_time"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) last_time_str = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') except Exception as e: print(f"Warning: Could not parse last_{ep_name}_time: {e}") if last_time_str is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) last_time_str = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') end_time_str = now.strftime('%Y-%m-%dT%H:%M:%SZ') print(f"Fetching {ep_name} from {last_time_str} to {end_time_str}") records, newest_event_time = fetch_logs( api_base=api_base, token=token, endpoint=endpoint, start_time=last_time_str, end_time=end_time_str, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) # Tag records with endpoint type for record in records: record['_lw_log_type'] = ep_name all_records.extend(records) # Update state for this endpoint if newest_event_time: state[f"last_{ep_name}_time"] = newest_event_time else: state[f"last_{ep_name}_time"] = end_time_str print(f"Fetched {len(records)} {ep_name} records") if not all_records: print("No new log records found.") save_state(bucket, STATE_KEY, state) 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}") # Save state save_state(bucket, STATE_KEY, state) print(f"Successfully processed {len(all_records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise 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) 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, state: dict): """Save the state to GCS state file.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: {json.dumps(state)}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(api_base: str, token: str, endpoint: dict, start_time: str, end_time: str, page_size: int, max_records: int): """ Fetch logs from Lacework API with pagination and rate limiting. Args: api_base: API base URL token: Bearer access token endpoint: Endpoint configuration dict start_time: Start time in ISO format end_time: End time in ISO format page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-LaceworkCollector/1.0' } ep_path = endpoint['path'] time_field = endpoint['time_field'] results_key = endpoint['results_key'] records = [] newest_time = None page_num = 0 backoff = 1.0 next_page = None while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records}) for {endpoint['name']}") break # Build request URL if next_page: url = next_page else: url = f"{api_base}{ep_path}?startTime={start_time}&endTime={end_time}" 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 [], None data = json.loads(response.data.decode('utf-8')) page_results = data.get(results_key, []) if not page_results: print(f"No more results (empty page) for {endpoint['name']}") break print(f"Page {page_num}: Retrieved {len(page_results)} {endpoint['name']} events") records.extend(page_results) # Track newest event time for event in page_results: try: event_time = event.get(time_field) if event_time: 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 next page via paging object paging = data.get('paging', {}) next_page_url = paging.get('urls', {}).get('nextPage') if not next_page_url: print(f"No more pages for {endpoint['name']}") break next_page = next_page_url except Exception as e: print(f"Error fetching {endpoint['name']} logs: {e}") return [], None print(f"Retrieved {len(records)} total {endpoint['name']} records from {page_num} pages") return records, newest_timerequirements.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」。
點選「建立工作」。
請提供下列設定詳細資料：
設定 值 名稱 lacework-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 lacework-logs-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率：
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
大量低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
少量、批次處理 |
| 每日 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中找出您的工作。
- 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下
lacework-logs-collector。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找：
Successfully obtained access token Fetching alerts from YYYY-MM-DDTHH:MM:SSZ to YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X alerts events Fetched X alerts records Fetching audit_logs from YYYY-MM-DDTHH:MM:SSZ to YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X audit_logs events Fetched X audit_logs records Wrote X records to gs://lacework-logs/lacework/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (
lacework-logs)。前往
lacework/資料夾。確認是否已建立新的
.ndjson檔案，且檔案名稱包含目前的時間戳記。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 API 憑證，或權杖可能已過期
- HTTP 403：在 Lacework 控制台中確認 API 金鑰是否具備必要權限
- HTTP 429：頻率限制 - 函式會自動重試，並延遲一段時間
- 缺少環境變數：檢查是否已設定所有必要變數
在 Google SecOps 中設定動態消息，擷取 FortiCNAPP (舊稱 Lacework) 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Lacework Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Lacework Cloud Security」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
gs://lacework-logs/lacework/- 取代：
lacework-logs：您的 GCS bucket 名稱。lacework：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。
- 取代：
來源刪除選項：根據偏好設定選取刪除選項：
- 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)
資產命名空間：資產命名空間
擷取標籤：要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 點按 bucket 名稱。
- 前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色：選取「Storage 物件檢視者」
按一下 [儲存]。
支援的 Lacework Cloud Security 範例記錄
代理程式或機器資訊 (主機清單)
{ "AGENT_VERSION": "6.7.6-4ce73a7b", "CREATED_TIME": "Thu, 03 Nov 2022 02:09:36 -0700", "HOSTNAME": "host-agent-1", "IP_ADDR": "10.0.0.1", "LAST_UPDATE": "Wed, 18 Oct 2023 17:59:09 -0700", "MID": 6516601498285932156, "MODE": "ebpf", "OS": "Linux", "STATUS": "ACTIVE", "TAGS": { "Account": "999999999999", "AmiId": "ami-00000000000000000", "ExternalIp": "203.0.113.10", "Hostname": "internal-host-1.zone.compute.internal", "InstanceId": "i-00000000000000000", "InternalIp": "172.16.1.10", "LwTokenShort": "DUMMYTOKENABCD123456", "Name": "proxy-DMZ-app-1", "ResourceType": "proxy-machines", "SubnetId": "subnet-00000000000000000", "VmInstanceType": "t3.small", "VmProvider": "AWS", "VpcId": "vpc-00000000000000000", "Zone": "us-west-2a", "arch": "amd64", "falconx.io/application": "proxy-machines", "falconx.io/environment": "prod", "falconx.io/project": "edge", "falconx.io/team": "edge", "os": "linux" } }檔案中繼資料或完整性
{ "CREATED_TIME": "Wed, 18 Oct 2023 17:02:01 -0700", "FILEDATA_HASH": "DUMMYHASH582C741AD91CA817B4718DEAA4E8A83C0B9D92E2", "FILE_PATH": "/usr/local/bin/secure_config", "MID": 7371220731851617371, "MTIME": "Fri, 25 Aug 2023 13:03:09 -0700", "SIZE": 8078 }主機安全漏洞評估
{ "CVE_PROPS": { "description": "DOCUMENTATION: The MITRE CVE dictionary describes this issue as: " "This CVE ID has been rejected or withdrawn by its CVE Numbering " "Authority for the following reason: This CVE ID has been rejected " "or withdrawn by its CVE Numbering Authority.", "link": "https://vendor.example.com/security/cve/CVE-2021-47472", "metadata": null }, "CVE_RISK_INFO": { "HOST_COUNT": 1249, "IMAGE_COUNT": 0, "PKG_COUNT": 0, "SEVERITY_LEVEL": 2, "score": 0.5154245281584533 }, "CVE_RISK_SCORE": 3.77, "END_TIME": "2024-09-04 07:00:00.000", "EVAL_CTX": { "collector_type": "Agent", "exception_props": [], "hostname": "vuln-host-1.example.net" }, "EVAL_GUID": "3dc61df780e3b722aa59b0ffcac85683", "FEATURE_KEY": { "name": "kernel-headers", "namespace": "centos:7", "package_active": 1, "package_path": "", "version_installed": "0:3.10.0-1160.119.1.el7.tuxcare.els2" }, "MACHINE_TAGS": { "Account": "999999999999", "AmiId": "ami-00000000000000000", "ExternalIp": "203.0.113.10", "Hostname": "ip-172-16-1-10.example-prod.aws.featurespace.net", "InternalIp": "10.0.0.1", "LwTokenShort": "DUMMYTOKENABCD123456", "VmProvider": "AWS", "VpcId": "vpc-00000000000000000", "os": "linux" }, "MID": 5746003737030963813, "PACKAGE_STATUS": "ACTIVE", "REGION": "eu-west-2", "RISK_SCORE": 10, "SEVERITY": "Low", "START_TIME": "2024-09-04 06:00:00.000", "STATUS": "Exception", "VULN_ID": "CVE-2021-47472" }雲端設定法規遵循 (稽核)
{ "ACCOUNT": { "AccountId": "999999999999", "Account_Alias": "" }, "EVAL_TYPE": "LW_SA", "ID": "lacework-global-87", "REASON": "Default security group does not restrict traffic", "RECOMMENDATION": "Ensure the default security group of every Virtual Private Cloud (VPC) restricts all traffic", "REGION": "eu-north-1", "REPORT_TIME": "2024-11-10 18:00:00.000", "RESOURCE_ID": "arn:aws:ec2:eu-west-1:999999999999:security-group/sg-00000000000000000", "SECTION": "", "SEVERITY": "High", "STATUS": "NonCompliant" }DNS 查詢或解析
{ "CREATED_TIME": "2024-11-06 05:14:44.329", "DNS_SERVER_IP": "10.0.0.53", "FQDN": "data-service-prod-1234567890.s3.eu-west-2.amazonaws.com", "HOST_IP_ADDR": "172.16.1.20", "MID": 8843985456817096491, "TTL": 5 }映像檔安全漏洞評估
{ "CVE_PROPS": null, "EVAL_CTX": { "collector_type": "Agentless", "image_info": { "digest": "sha256:52d5cb782dad7a8a03c8bd1b285bbd32bdbfa8fcc435614bb1e6ceefcf26ae1d", "id": "sha256:31427c44cac7ab632d541181073bbd46a964e4ed38d087d8a47f60bb66eef4df", "registry": "999999999999.dkr.ecr.eu-west-1.amazonaws.com", "repo": "amazon/aws-network-policy-agent" } }, "EVAL_GUID": "3a17a74f0a65eed2bddd2d37bb02e6af", "FEATURE_KEY": { "name": "perl-threads", "namespace": "amzn:2", "version": "1.87-4.amzn2.0.2" }, "FIX_INFO": { "fix_available": 0, "fixed_version": "" }, "IMAGE_ID": "sha256:31427c44cac7ab632d541181073bbd46a964e4ed38d087d8a47f60bb66eef4df", "IMAGE_RISK_INFO": { "factors": [ "cve", "reachability" ], "factors_breakdown": { "cve_counts": { "Critical": 0, "High": 21, "Medium": 73 }, "internet_reachability": "Unknown" } }, "IMAGE_RISK_SCORE": 6.4, "PACKAGE_STATUS": "NO_AGENT_AVAILABLE", "RISK_SCORE": 6.4, "START_TIME": "2024-11-05 19:05:03.553", "STATUS": "GOOD" }網路流量或連線摘要
{ "DST_ENTITY_ID": { "hostname": "service-A.region.amazonaws.com", "ip_internal": 0, "port": 443, "protocol": "TCP" }, "DST_ENTITY_TYPE": "DnsSep", "DST_IN_BYTES": 0, "DST_OUT_BYTES": 0, "ENDPOINT_DETAILS": [ { "dst_ip_addr": "203.0.113.10", "dst_port": 443, "protocol": "TCP", "src_ip_addr": "192.168.1.10" }, { "dst_ip_addr": "198.51.100.5", "dst_port": 443, "protocol": "TCP", "src_ip_addr": "192.168.1.10" } ], "END_TIME": "2024-11-05 21:00:00.000", "NUM_CONNS": 4, "SRC_ENTITY_ID": { "mid": 2080882850610892909, "pid_hash": 744766973756676842 }, "SRC_ENTITY_TYPE": "Process", "SRC_IN_BYTES": 25028, "SRC_OUT_BYTES": 11962, "START_TIME": "2024-11-05 20:00:00.000" }包裹資訊或最新消息
{ "ARCH": "x86_64", "CREATED_TIME": "2024-11-08 01:28:30.566", "MID": 4172267319977985370, "PACKAGE_NAME": "grub2", "VERSION": "2:2.02-0.87.0.2.el7.el7.centos.14.tuxcare.els2" }容器程序活動
{ "CONTAINER_ID": "4853339865add970f72213ec5d76ff51d1308c61a7680cc23c8de20c38c0a8e1", "END_TIME": "2024-11-08 02:00:00.000", "FILE_PATH": "/app/grpc-health-probe", "MID": 3708952045169222383, "PID": 177267, "POD_NAME": "kubernetes-pod-abc", "PPID": 177257, "PROCESS_START_TIME": "2024-11-08 01:43:29.960", "START_TIME": "2024-11-08 01:00:00.000", "UID": 0, "USERNAME": "serviceuser" }一般警報或事件 (CloudTrail)
{ "EVENT_ID": "413328", "EVENT_NAME": "Unauthorized API Call", "EVENT_TYPE": "CloudTrailDefaultAlert", "SUMMARY": " For account: 999999999999 (and 22 more) : event Unauthorized API Call from a username other " "than whitelisted ones. Replaces lacework-global-29 occurred 3772 times by user " "UDM-PRINCIPAL-ID:UDM-SERVICE-ROLE (and 167 more) ", "START_TIME": "07 Feb 2025 12:00 GMT", "EVENT_CATEGORY": "Aws", "LINK": "https://security.example.net/ui/alert/12345/details", "ACCOUNT": "UDM_ACCOUNT", "SOURCE": "CloudTrail", "subject": { "srcEvent": { "event": { "errorCode": "AccessDenied", "errorMessage": "User: arn:aws:sts::999999999999:assumed-role/UDM-SERVICE-ROLE-IngestionApiRole/UDM-SERVICE-PRINCIPAL " "is not authorized to perform: kinesis:ListShards on resource: " "arn:aws:kinesis:us-east-1:999999999999:stream/ingestion-qa-rel-fraud-review-Stream " "because no identity-based policy allows the kinesis:ListShards action", "eventName": "ListShards", "eventSource": "kinesis.amazonaws.com", "eventTime": "2025-02-07T12:00:24Z", "recipientAccountId": "999999999999", "sourceIPAddress": "firehose.amazonaws.com", "userIdentity": { "accessKeyId": "ACCESSKEYIDDUMMY", "accountId": "999999999999", "arn": "arn:aws:sts::999999999999:assumed-role/UDM-SERVICE-ROLE-IngestionApiRole/UDM-SERVICE-PRINCIPAL", "sessionContext": { "sessionIssuer": { "accountId": "999999999999", "arn": "arn:aws:iam::999999999999:role/UDM-SERVICE-ROLE-IngestionApiRole", "principalId": "PRINCIPALIDDUMMY", "userName": "UDM-SERVICE-ROLE-IngestionApiRole" } } }, "vpcEndpointId": "vpce-00000000000000000" }, "principalId": "PRINCIPALIDDUMMY:UDM-SERVICE-PRINCIPAL", "recipientAccountId": "999999999999", "sourceIPAddress": "firehose.amazonaws.com", "userIdentityName": "UDM-SERVICE-ROLE-IngestionApiRole" } } }
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| alertId | metadata.product_log_id | 直接複製值 |
| alertName | security_result.rule_name | 直接複製值 |
| 嚴重性 | security_result.severity | 對應至 UDM 嚴重性 |
| 狀態 | security_result.summary | 直接複製值 |
| alertType | security_result.category_details | 直接複製值 |
| startTime | metadata.event_timestamp | 剖析為 ISO 8601 時間戳記 |
| endTime | additional.fields | 儲存為 end_time 標籤 |
| alertInfo.description | security_result.description | 直接複製值 |
| alertInfo.subject | metadata.description | 直接複製值 |
| entityMap.Machine.hostname | principal.hostname | 直接複製值 |
| entityMap.Machine.externalIp | principal.ip | 直接複製值 |
| entityMap.User.username | principal.user.userid | 直接複製值 |
| entityMap.Region.region | principal.location.name | 直接複製值 |
| entityMap.CT_User.accountId | principal.user.product_object_id | 直接複製值 |
event_title |
event.idm.read_only_udm.security_result.summary |
從變更記錄對應 |
event_description |
event.idm.read_only_udm.security_result.description |
從變更記錄對應 |
summary_details |
event.idm.read_only_udm.security_result.confidence_details |
從變更記錄對應 |
target_user_id |
event.idm.read_only_udm.target.user.userid |
從變更記錄對應 |
resource_product_id |
event.idm.read_only_udm.target.resource.product_object_id |
從變更記錄對應 |
user_role |
event.idm.read_only_udm.target.user.role_name |
從變更記錄對應 |
account_name |
event.idm.read_only_udm.target.resource.name |
從變更記錄對應 |
application |
event.idm.read_only_udm.target.application |
從變更記錄對應 |
event_id |
event.idm.read_only_udm.metadata.product_log_id |
從變更記錄對應 |
event_type |
event.idm.read_only_udm.metadata.product_event_type |
從變更記錄對應 |
event_severity |
event.idm.read_only_udm.security_result.severity_details |
從變更記錄對應 |
lacework_account |
event.idm.read_only_udm.principal.user.userid |
從變更記錄對應 |
event_link |
event.idm.read_only_udm.metadata.url_back_to_product |
從變更記錄對應 |
starttimevalue |
event.idm.read_only_udm.additional.fields |
從變更記錄對應 |
endtimevalue |
event.idm.read_only_udm.additional.fields |
從變更記錄對應 |
intgGuid |
event.idm.read_only_udm.target.resource.attribute.labels |
從變更記錄對應 |
rec_id |
event.idm.read_only_udm.target.resource.attribute.labels |
從變更記錄對應 |
event_source |
event.idm.read_only_udm.metadata.product_name |
從變更記錄對應 |
event_timestamp |
event.idm.read_only_udm.metadata.event_timestamp |
從變更記錄對應 |
ACCOUNT", "EVENT_CATEGORY", "subject.srcEvent.recipientAccountAlias", "DERIVED_FIELDS.SOURCE", "subject.srcEvent.event.userIdentity.accessKeyId", "subject.srcEvent.event.userIdentity.arn", "subject.srcEvent.event.errorCode", "subject.srcEvent.event.errorMessage", "subject.srcEvent.event.eventID", "subject.srcEvent.event.eventSource", "subject.srcEvent.event.userIdentity.sessionContext.attributes.mfaAuthenticated", "subject.srcEvent.username", "subject.startTime", "subject.srcEvent.eventName", "DERIVED_FIELDS.CATEGORY", "DERIVED_FIELDS.SUBCATEGORY", "subject.dstEvent.gbm_version", "subject.dstEvent.is_visible", "subject.dstEvent.severity", "subject.dstEvent.recipientAccountAlias", "subject.srcEvent.api", "subject.srcEvent.calltype", "subject.srcEvent.gbm_version", "subject.srcEvent.is_visible", and "subject.srcEvent.severity |
additional.fields |
從變更記錄對應 |
SUMMARY |
metadata.description |
從變更記錄對應 |
EVENT_TYPE |
metadata.product_event_type |
從變更記錄對應 |
EVENT_ID |
metadata.product_log_id |
從變更記錄對應 |
LINK |
metadata.url_back_to_product |
從變更記錄對應 |
subject.srcEvent.event.userAgent", "subject.srcEvent.source |
network.http.user_agent |
從變更記錄對應 |
subject.srcEvent.recipientAccountId |
principal.user.groupid |
從變更記錄對應 |
subject.srcEvent.principalId |
principal.user.userid |
從變更記錄對應 |
subject.srcEvent.event.awsRegion |
security_result.about.asset.attribute.cloud.availability_zone |
從變更記錄對應 |
subject.srcEvent.event.eventCategory |
security_result.about.asset.category |
從變更記錄對應 |
EVENT_NAME |
security_result.category |
從變更記錄對應 |
EVENT_NAME |
security_result.summary |
從變更記錄對應 |
subject.srcType |
src.resource.resource_subtype |
從變更記錄對應 |
subject.srcEvent.event.userIdentity.sessionContext.attributes.creationDate |
metadata.event_timestamp |
從變更記錄對應 |
subject.srcEvent.accountcaller |
principal.resource.product_object_id |
從變更記錄對應 |
subject.dstEvent.region |
target.asset.location.name |
從變更記錄對應 |
subject.dstEvent.accountcaller |
target.resource.product_object_id |
從變更記錄對應 |
subject.dstType |
target.resource.resource_subtype |
從變更記錄對應 |
subject.dstEvent.service |
target.url |
從變更記錄對應 |
subject.dstEvent.username |
target.user.userid |
從變更記錄對應 |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。