收集 Mandiant Threat Intelligence 自訂 IOC 記錄
本文說明如何使用 Google Cloud Storage V2,將 Mandiant Threat Intelligence 自訂 IOC 資料擷取至 Google Security Operations。
Mandiant Threat Intelligence 提供豐富的入侵指標 (IOC),包括 IP 位址、網域、網址、檔案雜湊和電子郵件地址。這個平台會根據第一線事件應變調查提供威脅情報,協助安全團隊找出並優先處理針對貴機構的威脅。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 有效的 Mandiant Advantage Threat Intelligence 訂閱方案
- 享有 Mandiant Advantage 入口網站的專屬存取權
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 mandiant-ioc-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或資料保留政策 點選「建立」。
收集 Mandiant Advantage 憑證
取得 Mandiant 網站網址
- 登入 Mandiant Advantage 執行個體。
- 記下瀏覽器網址列中的網站網址。
- 網址為:
https://advantage.mandiant.com
- 網址為:
建立 API 憑證
- 登入 Mandiant Advantage。
- 前往 [設定]。
- 向下捲動至「API Access and Keys」(API 存取權和金鑰),或從導覽選單中選取該選項。
- 按一下「取得金鑰 ID 和密鑰」。
- 複製並安全地儲存 API 憑證:
- 金鑰 ID:複製這個值 (也稱為 API 金鑰 ID 或公開金鑰)
- 金鑰密鑰:複製這個值 (也稱為 API 密鑰或私密金鑰)
驗證權限
如要確認帳戶是否具備必要權限,請按照下列步驟操作:
- 登入 Mandiant Advantage。
- 前往 [設定]。
- 查看「使用者資訊」下的使用者權限。
- 確認您有權存取「威脅情報」模組。
- 如果無法查看威脅情報資料,請與 Mandiant 管理員聯絡,請對方授予適當的權限。
測試 API 存取權
請先測試憑證,再繼續進行整合:
# Replace with your actual credentials API_KEY_ID="your-key-id" API_SECRET="your-key-secret" API_BASE="https://api.intelligence.mandiant.com" # Get OAuth token TOKEN=$(curl -s -X POST "${API_BASE}/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials" \ -u "${API_KEY_ID}:${API_SECRET}" | jq -r '.access_token') # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" \ "${API_BASE}/v4/indicator?limit=1"
如果測試成功,您應該會收到含有指標資料的 JSON 回應。如果收到錯誤訊息,請按照下列步驟操作:
- HTTP 401:請檢查 API 金鑰 ID 和密鑰是否正確
- HTTP 403:確認帳戶是否具備威脅情報權限
- HTTP 404:確認 API 基準網址是否正確
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限,且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
mandiant-ioc-collector-sa(例如mandiant-ioc-collector-sa) - 服務帳戶說明:輸入
Service account for Cloud Run function to collect Mandiant IOC 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 名稱 (例如
mandiant-ioc-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
mandiant-ioc-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
mandiant-ioc-trigger(例如mandiant-ioc-trigger) - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Mandiant Threat Intelligence API 擷取 IOC 資料,並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 mandiant-ioc-collector(例如mandiant-ioc-collector)區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇 Pub/Sub 主題 (例如
mandiant-ioc-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
- 向下捲動並展開「容器」、「網路」和「安全性」。
- 前往「安全性」分頁:
- 服務帳戶:選取服務帳戶 (例如
mandiant-ioc-collector-sa)
- 服務帳戶:選取服務帳戶 (例如
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETmandiant-ioc-logsGCS bucket 名稱 GCS_PREFIXmandiant-ioc記錄檔的前置字串 STATE_KEYmandiant-ioc/state.json狀態檔案路徑 API_BASEhttps://api.intelligence.mandiant.comAPI 基準網址 API_KEY_IDyour-key-idMandiant API 金鑰 ID API_SECRETyour-key-secretMandiant API 密鑰 MAX_RECORDS1000每次執行的記錄數上限 PAGE_SIZE100每頁記錄數 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 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mandiant-ioc') STATE_KEY = os.environ.get('STATE_KEY', 'mandiant-ioc/state.json') API_BASE = os.environ.get('API_BASE', 'https://api.intelligence.mandiant.com') API_KEY_ID = os.environ.get('API_KEY_ID') API_SECRET = os.environ.get('API_SECRET') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) def to_unix_seconds(dt: datetime) -> int: """Convert datetime to Unix epoch seconds.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return int(dt.timestamp()) 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 Mandiant IOC data and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, API_BASE, API_KEY_ID, API_SECRET]): print('Error: Missing required environment variables') return try: # Get GCS bucket 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 IOCs from {last_time.isoformat()} to {now.isoformat()}") # Convert to Unix timestamps start_epoch = to_unix_seconds(last_time) end_epoch = to_unix_seconds(now) # Fetch IOCs records, newest_event_time = fetch_iocs( api_base=API_BASE, api_key_id=API_KEY_ID, api_secret=API_SECRET, start_epoch=start_epoch, end_epoch=end_epoch, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) if not records: print("No new IOC 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}/iocs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(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(records)} records") except Exception as e: print(f'Error processing IOCs: {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 get_oauth_token(api_base: str, api_key_id: str, api_secret: str) -> str: """Get OAuth 2.0 access token from Mandiant API.""" token_url = f"{api_base}/token" # Create Basic Auth header auth_string = f"{api_key_id}:{api_secret}" auth_bytes = auth_string.encode('utf-8') auth_b64 = base64.b64encode(auth_bytes).decode('utf-8') headers = { 'Authorization': f'Basic {auth_b64}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } body = 'grant_type=client_credentials' try: response = http.request('POST', token_url, body=body, headers=headers) if response.status != 200: print(f"Token request failed: {response.status}") print(f"Response: {response.data.decode('utf-8')}") raise Exception(f"Failed to get OAuth token: HTTP {response.status}") token_data = json.loads(response.data.decode('utf-8')) return token_data['access_token'] except Exception as e: print(f"Error getting OAuth token: {e}") raise def fetch_iocs(api_base: str, api_key_id: str, api_secret: str, start_epoch: int, end_epoch: int, page_size: int, max_records: int): """ Fetch IOCs from Mandiant Threat Intelligence API with pagination and rate limiting. Args: api_base: API base URL api_key_id: Mandiant API Key ID api_secret: Mandiant API Secret start_epoch: Start time in Unix epoch seconds end_epoch: End time in Unix epoch seconds page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ # Get OAuth token access_token = get_oauth_token(api_base, api_key_id, api_secret) endpoint = f"{api_base}/v4/indicator" headers = { 'Authorization': f'Bearer {access_token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-MandiantCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 next_url = None while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break # Build request URL if next_url: # Use next URL from pagination url = next_url else: # First request with time filtering params = [] params.append(f"start_epoch={start_epoch}") params.append(f"end_epoch={end_epoch}") params.append(f"limit={min(page_size, max_records - len(records))}") url = f"{endpoint}?{'&'.join(params)}" 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')) # Extract results page_results = data.get('indicators', []) if not page_results: print(f"No more results (empty page)") break print(f"Page {page_num}: Retrieved {len(page_results)} IOCs") records.extend(page_results) # Track newest event time for ioc in page_results: try: # Extract last_updated timestamp event_time = ioc.get('last_updated') 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 next_url = data.get('next') if not next_url: print("No more pages (no next URL)") break except Exception as e: print(f"Error fetching IOCs: {e}") return [], None print(f"Retrieved {len(records)} total 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」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 mandiant-ioc-collector-hourly(例如mandiant-ioc-collector-hourly)區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 (例如 mandiant-ioc-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據 IOC 更新量和延遲要求選擇頻率:
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *大量低延遲 每 15 分鐘 */15 * * * *中等 每小時 0 * * * *標準 (建議) 每 6 小時 0 */6 * * *少量、批次處理 每日 0 0 * * *歷來資料集合
測試整合項目
- 在 Cloud Scheduler 控制台中,找出您的工作 (例如
mandiant-ioc-collector-hourly)。 - 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱 (例如
mandiant-ioc-collector)。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找:
Fetching IOCs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X IOCs Wrote X records to gs://bucket-name/mandiant-ioc/iocs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
點按 bucket 名稱 (例如
mandiant-ioc-logs)。前往前置字串資料夾 (例如
mandiant-ioc/)。確認是否已建立新的
.ndjson檔案,且檔案名稱包含目前的時間戳記。
如果記錄中顯示錯誤:
- HTTP 401:檢查環境變數中的 API 金鑰 ID 和密鑰
- HTTP 403:確認帳戶是否具備威脅情報權限
- HTTP 429:頻率限制 - 函式會自動重試,並延遲一段時間
- 缺少環境變數:檢查是否已設定所有必要變數
- 權杖要求失敗:確認 API 憑證正確無誤且未過期
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Mandiant Custom IOC)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Mandiant Custom IOC」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://mandiant-ioc-logs/mandiant-ioc/- 取代:
mandiant-ioc-logs:您的 GCS bucket 名稱。mandiant-ioc:儲存 IOC 記錄的前置字元/資料夾路徑。
- 取代:
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)
資產命名空間:資產命名空間
擷取標籤:要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 點按 bucket 名稱 (例如
mandiant-ioc-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色:選取「Storage 物件檢視者」
- 按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| associated_hashes.value | entity.file.md5 | 如果 hash.type ==「md5」,則值來自 associated_hashes.value;否則,如果 type ==「md5」,則值來自 value |
| 值 | entity.file.md5 | |
| associated_hashes.value | entity.file.sha1 | 如果 hash.type ==「sha1」,則值來自 associated_hashes.value;如果 type ==「sha1」,則值來自 value |
| 值 | entity.file.sha1 | |
| associated_hashes.value | entity.file.sha256 | 如果 hash.type ==「sha256」,則為 associated_hashes.value 中的值;如果 type ==「sha256」,則為 value 中的值 |
| 值 | entity.file.sha256 | |
| 值 | entity.hostname | 如果 type == "fqdn",則直接複製值 |
| 值 | entity.ip | 如果 type == "ipv4",則直接複製值 |
| threat_rating.threat_score | entity.labels | 合併的標籤物件,包含「threat_score」鍵和 threat_rating.threat_score 的值、「mscore」鍵和 mscore 的值、「is_exclusive」鍵和 is_exclusive 的值、「is_publishable」鍵和 ispublishable 的值,以及每個 misp 欄位的「misp |
| mscore | entity.labels | |
| is_exclusive | entity.labels | |
| is_publishable | entity.labels | |
| misp.akamai | entity.labels | |
| misp.alexa | entity.labels | |
| misp.alexa_1M | entity.labels | |
| misp.amazon-aws | entity.labels | |
| misp.apple | entity.labels | |
| misp.automated-malware-analysis | entity.labels | |
| misp.bank-website | entity.labels | |
| misp.captive-portals | entity.labels | |
| misp.censys-scanning | entity.labels | |
| misp.cisco_1M | entity.labels | |
| misp.cisco_top1000 | entity.labels | |
| misp.cisco_top10k | entity.labels | |
| misp.cisco_top20k | entity.labels | |
| misp.cisco_top5k | entity.labels | |
| misp.cloudflare | entity.labels | |
| misp.common-contact-emails | entity.labels | |
| misp.common-ioc-false-positive | entity.labels | |
| misp.covid | entity.labels | |
| misp.covid-19-cyber-threat-coalition-whitelist | entity.labels | |
| misp.covid-19-krassi-whitelist | entity.labels | |
| misp.crl-hostname | entity.labels | |
| misp.crl-ip | entity.labels | |
| misp.dax30 | entity.labels | |
| misp.digitalside | entity.labels | |
| misp.disposable-email | entity.labels | |
| misp.dynamic-dns | entity.labels | |
| misp.eicar.com | entity.labels | |
| misp.empty-hashes | entity.labels | |
| misp.fastly | entity.labels | |
| misp.findip-host | entity.labels | |
| misp.google | entity.labels | |
| misp.google-chrome-crux-1million | entity.labels | |
| misp.google-gcp | entity.labels | |
| misp.google-gmail-sending-ips | entity.labels | |
| misp.googlebot | entity.labels | |
| misp.ipv6-linklocal | entity.labels | |
| misp.majestic_million | entity.labels | |
| misp.majestic_million_1M | entity.labels | |
| misp.microsoft | entity.labels | |
| misp.microsoft-attack-simulator | entity.labels | |
| misp.microsoft-azure | entity.labels | |
| misp.microsoft-azure-appid | entity.labels | |
| misp.microsoft-azure-china | entity.labels | |
| misp.microsoft-azure-germany | entity.labels | |
| misp.microsoft-azure-us-gov | entity.labels | |
| misp.microsoft-office365 | entity.labels | |
| misp.microsoft-office365-cn | entity.labels | |
| misp.microsoft-office365-ip | entity.labels | |
| misp.microsoft-win10-connection-endpoints | entity.labels | |
| misp.moz-top500 | entity.labels | |
| misp.mozilla-CA | entity.labels | |
| misp.mozilla-IntermediateCA | entity.labels | |
| misp.multicast | entity.labels | |
| misp.nioc-filehash | entity.labels | |
| misp.openai-gptbot | entity.labels | |
| misp.ovh-cluster | entity.labels | |
| misp.parking-domain | entity.labels | |
| misp.parking-domain-ns | entity.labels | |
| misp.phone_numbers | entity.labels | |
| misp.public-dns-hostname | entity.labels | |
| misp.public-dns-v4 | entity.labels | |
| misp.public-dns-v6 | entity.labels | |
| misp.public-ipfs-gateways | entity.labels | |
| misp.rfc1918 | entity.labels | |
| misp.rfc3849 | entity.labels | |
| misp.rfc5735 | entity.labels | |
| misp.rfc6598 | entity.labels | |
| misp.rfc6761 | entity.labels | |
| misp.second-level-tlds | entity.labels | |
| misp.security-provider-blogpost | entity.labels | |
| misp.sinkholes | entity.labels | |
| misp.smtp-receiving-ips | entity.labels | |
| misp.smtp-sending-ips | entity.labels | |
| misp.stackpath | entity.labels | |
| misp.tenable-cloud-ipv4 | entity.labels | |
| misp.tenable-cloud-ipv6 | entity.labels | |
| misp.ti-falsepositives | entity.labels | |
| misp.tlds | entity.labels | |
| misp.tranco | entity.labels | |
| misp.tranco10k | entity.labels | |
| misp.umbrella-blockpage-hostname | entity.labels | |
| misp.umbrella-blockpage-v4 | entity.labels | |
| misp.umbrella-blockpage-v6 | entity.labels | |
| misp.university_domains | entity.labels | |
| misp.url-shortener | entity.labels | |
| misp.vpn-ipv4 | entity.labels | |
| misp.vpn-ipv6 | entity.labels | |
| misp.whats-my-ip | entity.labels | |
| misp.wikimedia | entity.labels | |
| misp.zscaler | entity.labels | |
| 值 | entity.url | 如果 type == "url",則直接複製值 |
| last_seen | event.ioc.active_timerange.end | 使用 ISO8601 從 last_seen 剖析的日期 |
| first_seen | event.ioc.active_timerange.start | 使用 ISO8601 從 first_seen 剖析的日期 |
| sources.0.category | event.ioc.categorization | 以半形逗號分隔,從 sources.0.category 陣列中加入 |
| threat_rating.threat_score | event.ioc.confidence_score | 直接複製值 |
| 值 | event.ioc.domain_and_ports.domain | 如果 type == "fqdn",則直接複製值 |
| event.ioc.feed_name | event.ioc.feed_name | 設為「Mandiant」 |
| 值 | event.ioc.ip_and_ports.ip_address | 如果 type == "ipv4",則轉換為 ipaddress |
| 類型 | metadata.entity_type | 如果 associated_hashes.type 位於 md5 sha1 sha256 中,則設為 FILE;如果 type == "fqdn",則設為 DOMAIN_NAME;如果 type == "ipv4",則設為 IP_ADDRESS;如果 type == "url",則設為 URL;如果 type 位於 md5 sha1 sha256 中,則設為 FILE;否則設為 UNKNOWN_ENTITYTYPE |
| associated_hashes.type | metadata.entity_type | |
| custom_ioc_expire_date | metadata.interval.end_time | 如果類型為 fqdn ipv4 url,則使用 ISO8601 從 custom_ioc_expire_date 剖析的日期 |
| first_seen | metadata.interval.start_time | 使用 ISO8601 或 UNIX 從 first_seen 剖析的日期 |
| id | metadata.product_entity_id | 直接複製值 |
| threat_det | metadata.threat | 從 threat_det 合併 |
| attributed_associations | threat_det.associations | 從 attributed_associations 和 associations 合併 threat_association 物件,並從 association.id、name、type、region_code.country_or_region、associated_actors、alias、role、first_reference_time、last_reference_time、industries_affected 設定欄位 |
| 關聯 | threat_det.associations | |
| campaigns | threat_det.campaigns | 從廣告活動陣列合併 campaign.name |
| category | threat_det.category_details | 從類別陣列合併 |
| threat_rating.confidence_level | threat_det.confidence | 如果相符程度低,請設為 LOW_CONFIDENCE;如果相符程度中等,請設為 MEDIUM_CONFIDENCE;如果相符程度高,請設為 HIGH_CONFIDENCE;否則請設為 UNKNOWN_CONFIDENCE |
| threat_rating.confidence_score | threat_det.confidence_details | 從 threat_rating.confidence_score 轉換為字串的值,或從 mscore (如果 threat_rating.confidence_score 為空) 轉換為字串的值 |
| mscore | threat_det.confidence_details | |
| first_seen | threat_det.first_discovered_time | 如果 first_seen 不為空,則為 first_seen 中的日期,否則為 firstSeen 中的日期,採用 ISO8601 或 UNIX 格式 |
| firstSeen | threat_det.first_discovered_time | |
| first_seen | threat_det.last_updated_time | 使用 ISO8601 或 UNIX 從 first_seen 剖析的日期 |
| threat_rating.threat_score | threat_det.risk_score | 如果不是空白,則從 threat_rating.threat_score 轉換為浮點數,否則從 mscore 轉換 |
| mscore | threat_det.risk_score | |
| threat_rating.severity_level | threat_det.severity | 如果嚴重程度為低,請設為 LOW;如果為中,請設為 MEDIUM;如果為高,請設為 HIGH;否則請設為 UNKNOWN_SEVERITY |
| threat_det.threat_feed_name | threat_det.threat_feed_name | 設為「Mandiant」 |
| 值 | threat_det.url_back_to_product | 如果類型為 fqdn ipv4 url md5,請設為「https://advantage.mandiant.com/search?query=%{value}」 |
| verdict_simple.timestamp | threat_det.verdict_info | 從 verdict_info_simple 合併,並在設定時間戳記的 verdict_time、惡意 verdict 的 verdict_response,以及 verdict_source 的 verdict_type 後合併 |
| verdict_simple.verdict | threat_det.verdict_info | |
| verdict_simple.verdict_source | threat_det.verdict_info | |
| metadata.product_name | metadata.product_name | 設為「MANDIANT_CUSTOM_IOC」 |
| metadata.vendor_name | metadata.vendor_name | 設為「MANDIANT_CUSTOM_IOC」 |
threat.threat_feed_name |
Mandiant |
從變更記錄對應 |
threat_rating.threat_score |
entity.entity.labels |
從變更記錄對應 |
threat_rating.severity_level |
threat.severity |
從變更記錄對應 |
threat_rating.confidence_level |
threat.confidence |
從變更記錄對應 |
verdict_simple.timestamp |
verdict_info.verdict_time |
從變更記錄對應 |
category |
threat.category_details |
從變更記錄對應 |
threat_rating.confidence_score |
threat.confidence_details |
從變更記錄對應 |
threat_rating.threat_score |
threat.risk_score |
從變更記錄對應 |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。