收集 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

  1. 前往 Google Cloud 控制台
  2. 選取專案或建立新專案。
  3. 在導覽選單中,依序前往「Cloud Storage」>「Bucket」
  4. 按一下「建立值區」
  5. 請提供下列設定詳細資料:

    設定
    為 bucket 命名 輸入全域不重複的名稱 (例如 mandiant-ioc-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用:啟用物件版本管理或資料保留政策
  6. 點選「建立」

收集 Mandiant Advantage 憑證

取得 Mandiant 網站網址

  1. 登入 Mandiant Advantage 執行個體。
  2. 記下瀏覽器網址列中的網站網址。
    • 網址為:https://advantage.mandiant.com

建立 API 憑證

  1. 登入 Mandiant Advantage
  2. 前往 [設定]
  3. 向下捲動至「API Access and Keys」(API 存取權和金鑰),或從導覽選單中選取該選項。
  4. 按一下「取得金鑰 ID 和密鑰」
  5. 複製並安全地儲存 API 憑證:
    • 金鑰 ID:複製這個值 (也稱為 API 金鑰 ID 或公開金鑰)
    • 金鑰密鑰:複製這個值 (也稱為 API 密鑰或私密金鑰)

驗證權限

如要確認帳戶是否具備必要權限,請按照下列步驟操作:

  1. 登入 Mandiant Advantage。
  2. 前往 [設定]
  3. 查看「使用者資訊」下的使用者權限。
  4. 確認您有權存取「威脅情報」模組。
  5. 如果無法查看威脅情報資料,請與 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 叫用的服務帳戶。

建立服務帳戶

  1. GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)
  2. 按一下「Create Service Account」(建立服務帳戶)
  3. 請提供下列設定詳細資料:
    • 服務帳戶名稱:輸入 mandiant-ioc-collector-sa (例如 mandiant-ioc-collector-sa)
    • 服務帳戶說明:輸入 Service account for Cloud Run function to collect Mandiant IOC logs
  4. 按一下「建立並繼續」
  5. 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
    1. 按一下「選擇角色」
    2. 搜尋並選取「Storage 物件管理員」
    3. 點選「+ 新增其他角色」
    4. 搜尋並選取「Cloud Run Invoker」
    5. 點選「+ 新增其他角色」
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)
  6. 按一下「繼續」
  7. 按一下 [完成]

這些角色是下列作業的必要條件: - Storage 物件管理員:將記錄寫入 GCS bucket,並管理狀態檔案 - Cloud Run 叫用者:允許 Pub/Sub 叫用函式 - Cloud Functions 叫用者:允許叫用函式

授予 GCS 值區的 IAM 權限

將 GCS bucket 的寫入權限授予服務帳戶:

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 點按 bucket 名稱 (例如 mandiant-ioc-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:輸入服務帳戶電子郵件地址 (例如 mandiant-ioc-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • 指派角色:選取「Storage 物件管理員」
  6. 按一下 [儲存]

建立 Pub/Sub 主題

建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。

  1. GCP Console 中,前往「Pub/Sub」>「Topics」(主題)
  2. 按一下「建立主題」
  3. 請提供下列設定詳細資料:
    • 主題 ID:輸入 mandiant-ioc-trigger (例如 mandiant-ioc-trigger)
    • 其他設定保留預設值
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Mandiant Threat Intelligence API 擷取 IOC 資料,並寫入 GCS。

  1. 前往 GCP Console 的「Cloud Run」
  2. 按一下「Create service」(建立服務)
  3. 選取「函式」 (使用內嵌編輯器建立函式)。
  4. 在「設定」部分,提供下列設定詳細資料:

    設定
    服務名稱 mandiant-ioc-collector (例如 mandiant-ioc-collector)
    區域 選取與 GCS bucket 相符的區域 (例如 us-central1)
    執行階段 選取 Python 3.12 以上版本
  5. 在「Trigger (optional)」(觸發條件 (選用)) 專區:

    1. 按一下「+ 新增觸發條件」
    2. 選取「Cloud Pub/Sub」
    3. 在「選取 Cloud Pub/Sub 主題」中,選擇 Pub/Sub 主題 (例如 mandiant-ioc-trigger)。
    4. 按一下 [儲存]
  6. 在「Authentication」(驗證) 部分:

    1. 選取「需要驗證」
    2. 檢查 Identity and Access Management (IAM)
  1. 向下捲動並展開「容器」、「網路」和「安全性」
  2. 前往「安全性」分頁:
    • 服務帳戶:選取服務帳戶 (例如 mandiant-ioc-collector-sa)
  3. 前往「容器」分頁:

    1. 按一下「變數與密鑰」
    2. 針對每個環境變數,按一下「+ 新增變數」
    變數名稱 範例值 說明
    GCS_BUCKET mandiant-ioc-logs GCS bucket 名稱
    GCS_PREFIX mandiant-ioc 記錄檔的前置字串
    STATE_KEY mandiant-ioc/state.json 狀態檔案路徑
    API_BASE https://api.intelligence.mandiant.com API 基準網址
    API_KEY_ID your-key-id Mandiant API 金鑰 ID
    API_SECRET your-key-secret Mandiant API 密鑰
    MAX_RECORDS 1000 每次執行的記錄數上限
    PAGE_SIZE 100 每頁記錄數
    LOOKBACK_HOURS 24 初始回溯期
  4. 在「變數與密鑰」部分,向下捲動至「要求」

    • 要求逾時:輸入 600 秒 (10 分鐘)
  5. 前往「設定」分頁:

    • 在「資源」部分:
      • 記憶體:選取 512 MiB 以上
      • CPU:選取 1
  6. 在「修訂版本資源調度」部分:

    • 執行個體數量下限:輸入 0
    • 執行個體數量上限:輸入 100 (或根據預期負載調整)
  7. 點選「建立」

  8. 等待服務建立完成 (1 到 2 分鐘)。

  9. 服務建立完成後,系統會自動開啟內嵌程式碼編輯器

新增函式程式碼

  1. 在「進入點」欄位中輸入「main」
  2. 在內嵌程式碼編輯器中,建立兩個檔案:

    • 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_time
      
    • requirements.txt:

      functions-framework==3.*
      google-cloud-storage==2.*
      urllib3>=2.0.0
      
  3. 點選「部署」即可儲存並部署函式。

  4. 等待部署作業完成 (2 到 3 分鐘)。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題,觸發 Cloud Run 函式。

  1. 前往 GCP 主控台的「Cloud Scheduler」
  2. 點選「建立工作」
  3. 請提供下列設定詳細資料:

    設定
    名稱 mandiant-ioc-collector-hourly (例如 mandiant-ioc-collector-hourly)
    區域 選取與 Cloud Run 函式相同的區域
    頻率 0 * * * * (每小時整點)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取 Pub/Sub 主題 (例如 mandiant-ioc-trigger)
    郵件內文 {} (空白 JSON 物件)
  4. 點選「建立」

排程頻率選項

  • 根據 IOC 更新量和延遲要求選擇頻率:

    頻率 Cron 運算式 用途
    每 5 分鐘 */5 * * * * 大量低延遲
    每 15 分鐘 */15 * * * * 中等
    每小時 0 * * * * 標準 (建議)
    每 6 小時 0 */6 * * * 少量、批次處理
    每日 0 0 * * * 歷來資料集合

測試整合項目

  1. Cloud Scheduler 控制台中,找出您的工作 (例如 mandiant-ioc-collector-hourly)。
  2. 按一下「強制執行」即可手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「Services」
  5. 按一下函式名稱 (例如 mandiant-ioc-collector)。
  6. 按一下 [Logs] (記錄) 分頁標籤。
  7. 確認函式是否已順利執行。尋找:

    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
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 點按 bucket 名稱 (例如 mandiant-ioc-logs)。

  10. 前往前置字串資料夾 (例如 mandiant-ioc/)。

  11. 確認是否已建立新的 .ndjson 檔案,且檔案名稱包含目前的時間戳記。

如果記錄中顯示錯誤:

  • HTTP 401:檢查環境變數中的 API 金鑰 ID 和密鑰
  • HTTP 403:確認帳戶是否具備威脅情報權限
  • HTTP 429:頻率限制 - 函式會自動重試,並延遲一段時間
  • 缺少環境變數:檢查是否已設定所有必要變數
  • 權杖要求失敗:確認 API 憑證正確無誤且未過期

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Mandiant Custom IOC)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「Mandiant Custom IOC」做為「記錄類型」
  7. 按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 複製這個電子郵件地址,以便在下一步中使用。

  9. 點選「下一步」

  10. 指定下列輸入參數的值:

    • 儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:

      gs://mandiant-ioc-logs/mandiant-ioc/
      
      • 取代:
        • mandiant-ioc-logs:您的 GCS bucket 名稱。
        • mandiant-ioc:儲存 IOC 記錄的前置字元/資料夾路徑。
    • 來源刪除選項:根據偏好設定選取刪除選項:

      • 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。

    • 檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)

    • 資產命名空間資產命名空間

    • 擷取標籤:要套用至這個動態饋給事件的標籤

  11. 點選「下一步」

  12. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 點按 bucket 名稱 (例如 mandiant-ioc-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
    • 指派角色:選取「Storage 物件檢視者」
  6. 按一下 [儲存]

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」鍵和 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 專業人員尋求答案。