收集 Censys 記錄

支援的國家/地區：

Google SecOps SIEM

注意： 這項功能受《Google Security Operations 服務專屬條款》的《正式發布前產品條款》規範。正式發布前的功能或許只提供部分支援，而且這類功能的變更可能與其他正式發布前版本不相容。詳情請參閱《Google SecOps 技術支援服務指南》和《Google SecOps 服務專屬條款》。

本文說明如何使用 Google Cloud Storage V2，將 Censys 記錄檔擷取至 Google Security Operations。

Censys 透過 API 提供全面的攻擊面管理和網際網路情報。這項整合功能可讓您從 Censys ASM 收集主機探索事件、風險事件和資產變更，並轉送至 Google SecOps 進行分析和監控。

事前準備

請確認您已完成下列事前準備事項：

• Google SecOps 執行個體
• 已啟用 Cloud Storage API 的 GCP 專案
• 建立及管理 GCS 值區的權限
• 管理 Google Cloud Storage 值區 IAM 政策的權限
• 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
• Censys ASM 的特殊存取權

收集 Censys API 憑證

1. 前往 app.censys.io 登入 Censys ASM 控制台。
2. 前往頁面頂端的「整合」。
3. 複製並儲存「API 金鑰」和「機構 ID」。
4. 請記下 API 基礎網址：https://api.platform.censys.io

建立 Google Cloud Storage 值區

1. 前往 Google Cloud 控制台。
2. 選取專案或建立新專案。
3. 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
4. 按一下「建立值區」。

5. 請提供下列設定詳細資料：

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

6. 點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶。

建立服務帳戶

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

這些角色適用於：

• Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
• Cloud Run 叫用者：允許 Pub/Sub 叫用函式
• Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

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

1. 依序前往「Cloud Storage」>「Buckets」。
2. 按一下 bucket 名稱。
3. 前往「權限」分頁標籤。
4. 按一下「授予存取權」。
5. 請提供下列設定詳細資料：
   • 新增主體：輸入服務帳戶電子郵件地址 (例如 censys-data-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：輸入 censys-data-trigger。
   • 其他設定保留預設值。
4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Censys ASM API 擷取記錄，並將記錄寫入 GCS。

1. 前往 GCP Console 的「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」 (使用內嵌編輯器建立函式)。

4. 在「設定」部分，提供下列設定詳細資料：

   設定	值
   服務名稱	censys-data-collector
   區域	選取與 GCS bucket 相符的區域 (例如 us-central1)
   執行階段	選取「Python 3.12」以上版本

5. 在「Trigger (optional)」(觸發條件 (選用)) 專區：

   1. 按一下「+ 新增觸發條件」。
   2. 選取「Cloud Pub/Sub」。
   3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (censys-data-trigger)。
   4. 按一下 [儲存]。

6. 在「Authentication」(驗證) 部分：

   • 選取「需要驗證」。
   • 檢查 Identity and Access Management (IAM)。

注意： 叫用函式時，Pub/Sub 會自動處理驗證。

1. 向下捲動並展開「Containers, Networking, Security」。
2. 前往「安全性」分頁：
   • 服務帳戶：選取服務帳戶 (censys-data-collector-sa)。

3. 前往「容器」分頁：

   1. 按一下「變數與密鑰」。

   2. 針對每個環境變數，按一下「+ 新增變數」：

      變數名稱	範例值
      GCS_BUCKET	censys-logs
      GCS_PREFIX	censys/
      STATE_KEY	censys/state.json
      CENSYS_API_KEY	your-censys-api-key
      CENSYS_ORG_ID	your-organization-id
      API_BASE	https://api.platform.censys.io

4. 在「變數與密鑰」分頁中向下捲動至「要求」：

   • 要求逾時：輸入 600 秒 (10 分鐘)。

5. 前往「容器」中的「設定」分頁：

   • 在「資源」部分：
     • 記憶體：選取 512 MiB 以上。
     • CPU：選取 1。
   • 按一下 [完成]。

6. 捲動至「執行環境」：

   • 選取「預設」 (建議選項)。

7. 在「修訂版本資源調度」部分：

   • 執行個體數量下限：輸入 0。
   • 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。

8. 點選「建立」。

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

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

新增函式程式碼

1. 在「Function entry point」(函式進入點) 中輸入 main

2. 在內嵌程式碼編輯器中建立兩個檔案：

   • 第一個檔案：main.py:

   import functions_framework
   from google.cloud import storage
   import json
   import urllib3
   import gzip
   import os
   from datetime import datetime, timedelta, timezone
   from typing import Dict, List, Any, Optional
   from urllib.parse import urlencode
   
   # 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()
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch logs from Censys ASM API and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       # Get environment variables
       bucket_name = os.environ.get('GCS_BUCKET')
       prefix = os.environ.get('GCS_PREFIX', 'censys/')
       state_key = os.environ.get('STATE_KEY', 'censys/state.json')
       censys_api_key = os.environ.get('CENSYS_API_KEY')
       censys_org_id = os.environ.get('CENSYS_ORG_ID')
       api_base = os.environ.get('API_BASE', 'https://api.platform.censys.io')
   
       if not all([bucket_name, censys_api_key, censys_org_id]):
           print('Error: Missing required environment variables')
           return
   
       try:
           collector = CensysCollector(
               bucket_name=bucket_name,
               prefix=prefix,
               state_key=state_key,
               api_key=censys_api_key,
               org_id=censys_org_id,
               api_base=api_base
           )
   
           # Get last collection time
           last_collection_time = collector.get_last_collection_time()
           current_time = datetime.now(timezone.utc)
   
           print(f'Collecting events since {last_collection_time}')
   
           # Collect different types of events
           logbook_events = collector.collect_logbook_events()
           risk_events = collector.collect_risks_events()
   
           # Save events to GCS
           collector.save_events_to_gcs(logbook_events, 'logbook')
           collector.save_events_to_gcs(risk_events, 'risks')
   
           # Update state
           collector.save_collection_time(current_time)
   
           print(f'Successfully processed {len(logbook_events)} logbook events and {len(risk_events)} risk events')
   
       except Exception as e:
           print(f'Error processing logs: {str(e)}')
           raise
   
   class CensysCollector:
       def __init__(self, bucket_name: str, prefix: str, state_key: str, 
                    api_key: str, org_id: str, api_base: str):
           self.bucket_name = bucket_name
           self.prefix = prefix
           self.state_key = state_key
           self.headers = {
               'Authorization': f'Bearer {api_key}',
               'X-Organization-ID': org_id,
               'Content-Type': 'application/json'
           }
           self.api_base = api_base
           self.bucket = storage_client.bucket(bucket_name)
   
       def get_last_collection_time(self) -> Optional[datetime]:
           """Get the last collection timestamp from GCS state file."""
           try:
               blob = self.bucket.blob(self.state_key)
               if blob.exists():
                   state_data = blob.download_as_text()
                   state = json.loads(state_data)
                   return datetime.fromisoformat(state.get('last_collection_time', '2024-01-01T00:00:00Z'))
           except Exception as e:
               print(f'No state file found or error reading state: {e}')
           return datetime.now(timezone.utc) - timedelta(hours=1)
   
       def save_collection_time(self, collection_time: datetime):
           """Save the current collection timestamp to GCS state file."""
           state = {'last_collection_time': collection_time.strftime('%Y-%m-%dT%H:%M:%SZ')}
           blob = self.bucket.blob(self.state_key)
           blob.upload_from_string(
               json.dumps(state),
               content_type='application/json'
           )
   
       def collect_logbook_events(self, cursor: str = None) -> List[Dict[str, Any]]:
           """Collect logbook events from Censys ASM API using cursor-based pagination."""
           events = []
           url = f"{self.api_base}/v3/logbook"
   
           params = {}
           if cursor:
               params['cursor'] = cursor
   
           try:
               query_string = urlencode(params) if params else ''
               full_url = f"{url}?{query_string}" if query_string else url
   
               response = http.request('GET', full_url, headers=self.headers)
   
               # Handle rate limiting with exponential backoff
               if response.status == 429:
                   retry_after = int(response.headers.get('Retry-After', '60'))
                   print(f'Rate limited (429). Retrying after {retry_after}s...')
                   import time
                   time.sleep(retry_after)
                   return self.collect_logbook_events(cursor)
   
               if response.status != 200:
                   print(f'API request failed with status {response.status}: {response.data}')
                   return []
   
               data = json.loads(response.data.decode('utf-8'))
               events.extend(data.get('logbook_entries', []))
   
               # Handle cursor-based pagination
               next_cursor = data.get('next_cursor')
               if next_cursor:
                   events.extend(self.collect_logbook_events(next_cursor))
   
               print(f'Collected {len(events)} logbook events')
               return events
   
           except Exception as e:
               print(f'Error collecting logbook events: {e}')
               return []
   
       def collect_risks_events(self) -> List[Dict[str, Any]]:
           """Collect risk events from Censys ASM API."""
           events = []
           url = f"{self.api_base}/v3/risks"
   
           try:
               response = http.request('GET', url, headers=self.headers)
   
               # Handle rate limiting with exponential backoff
               if response.status == 429:
                   retry_after = int(response.headers.get('Retry-After', '60'))
                   print(f'Rate limited (429). Retrying after {retry_after}s...')
                   import time
                   time.sleep(retry_after)
                   return self.collect_risks_events()
   
               if response.status != 200:
                   print(f'API request failed with status {response.status}: {response.data}')
                   return []
   
               data = json.loads(response.data.decode('utf-8'))
               events.extend(data.get('risks', []))
   
               print(f'Collected {len(events)} risk events')
               return events
   
           except Exception as e:
               print(f'Error collecting risk events: {e}')
               return []
   
       def save_events_to_gcs(self, events: List[Dict[str, Any]], event_type: str):
           """Save events to GCS in compressed NDJSON format."""
           if not events:
               return
   
           timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
           filename = f"{self.prefix}{event_type}_{timestamp}.json.gz"
   
           try:
               # Convert events to newline-delimited JSON
               ndjson_content = '\n'.join(json.dumps(event, separators=(',', ':')) for event in events)
   
               # Compress with gzip
               gz_bytes = gzip.compress(ndjson_content.encode('utf-8'))
   
               blob = self.bucket.blob(filename)
               blob.upload_from_string(
                   gz_bytes,
                   content_type='application/gzip'
               )
   
               print(f'Saved {len(events)} {event_type} events to {filename}')
   
           except Exception as e:
               print(f'Error saving {event_type} events to GCS: {e}')
               raise
   

   • 第二個檔案：requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. 點選「部署」來儲存並部署函式。

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

注意： Pub/Sub 觸發條件設定會自動建立必要的訂閱項目和權限。

建立 Cloud Scheduler 工作

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

1. 前往 GCP 主控台的「Cloud Scheduler」。
2. 點選「建立工作」。

3. 請提供下列設定詳細資料：

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

4. 點選「建立」。

排程頻率選項

• 根據記錄檔量和延遲時間要求選擇頻率：

  頻率	Cron 運算式	用途
  每 5 分鐘	*/5 * * * *	高容量、低延遲
  每 15 分鐘檢查一次	*/15 * * * *	普通量
  每小時	0 * * * *	標準 (建議採用)
  每 6 小時	0 */6 * * *	少量、批次處理
  每日	0 0 * * *	歷來資料集合

測試整合項目

1. 在 Cloud Scheduler 控制台中找出您的工作。
2. 按一下「強制執行」，手動觸發工作。
3. 稍等幾秒鐘。
4. 前往「Cloud Run」>「Services」。
5. 按一下函式名稱 (censys-data-collector)。
6. 按一下 [Logs] (記錄) 分頁標籤。

7. 確認函式是否已順利執行。請找出以下項目：

   Collecting events since YYYY-MM-DDTHH:MM:SS+00:00
   Collected X logbook events
   Collected X risk events
   Saved X logbook events to censys/logbook_YYYYMMDD_HHMMSS.json.gz
   Saved X risks events to censys/risks_YYYYMMDD_HHMMSS.json.gz
   Successfully processed X logbook events and X risk events
   

8. 依序前往「Cloud Storage」>「Buckets」。

9. 按一下 bucket 名稱。

10. 前往前置字元資料夾 (censys/)。

11. 確認是否已建立含有目前時間戳記的新 .json.gz 檔案。

如果在記錄中發現錯誤：

• HTTP 401：檢查環境變數中的 API 憑證
• HTTP 403：確認帳戶是否具備必要權限
• HTTP 429：頻率限制 - 函式會自動重試並延遲
• 缺少環境變數：檢查是否已設定所有必要變數

擷取 Google SecOps 服務帳戶

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

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

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態消息」。
3. 按一下「設定單一動態饋給」。
4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Censys logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「CENSYS」做為「記錄類型」。
7. 按一下「取得服務帳戶」。

8. 系統會顯示不重複的服務帳戶電子郵件地址，例如：

   [e-mail]
   

9. 複製這個電子郵件地址，以便在下一步中使用。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

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

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

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

注意： 如果您打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案和空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」。

在 Google SecOps 中設定動態饋給，擷取 Censys 記錄

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態消息」。
3. 按一下「設定單一動態饋給」。
4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Censys logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「CENSYS」做為「記錄類型」。
7. 點選 [下一步]。

8. 指定下列輸入參數的值：

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

     gs://censys-logs/censys/
     

     • 取代：

       • censys-logs：您的 GCS bucket 名稱。
       • censys/：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。

     • 範例：

       • 根層級 bucket：gs://censys-logs/
       • 前置字串：gs://censys-logs/censys/
     注意： URI 結尾一律要加上尾端斜線 (/)。

   • 來源刪除選項：根據偏好設定選取刪除選項：

     • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
     • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。

     • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。

       注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。

   • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。

   • 資產命名空間：資產命名空間。

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

9. 點選 [下一步]。

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

UDM 對應表

記錄欄位	UDM 對應	邏輯
assetId	read_only_udm.principal.asset.hostname	如果 assetId 欄位不是 IP 位址，則會對應至 principal.asset.hostname。
assetId	read_only_udm.principal.asset.ip	如果 assetId 欄位是 IP 位址，則會對應至 principal.asset.ip。
assetId	read_only_udm.principal.hostname	如果 assetId 欄位不是 IP 位址，則會對應至 principal.hostname。
assetId	read_only_udm.principal.ip	如果 assetId 欄位是 IP 位址，則會對應至 principal.ip。
associatedAt	read_only_udm.security_result.detection_fields.value	associatedAt 欄位會對應至 security_result.detection_fields.value。
autonomousSystem.asn	read_only_udm.additional.fields.value.string_value	autonomousSystem.asn 欄位會轉換為字串，並對應至 additional.fields.value.string_value，鍵為「autonomousSystem_asn」。
autonomousSystem.bgpPrefix	read_only_udm.additional.fields.value.string_value	autonomousSystem.bgpPrefix 欄位會對應至 additional.fields.value.string_value，並使用「autonomousSystem_bgpPrefix」鍵。
橫幅	read_only_udm.principal.resource.attribute.labels.value	橫幅欄位會對應至 principal.resource.attribute.labels.value，並以「banner」做為鍵。
雲朵	read_only_udm.metadata.vendor_name	雲端欄位會對應至 metadata.vendor_name。
comments.refUrl	read_only_udm.network.http.referral_url	comments.refUrl 欄位會對應至 network.http.referral_url。
data.cve	read_only_udm.additional.fields.value.string_value	data.cve 欄位會對應至 additional.fields.value.string_value，並使用「data_cve」鍵。
data.cvss	read_only_udm.additional.fields.value.string_value	data.cvss 欄位會對應至 additional.fields.value.string_value，並使用「data_cvss」鍵。
data.ipAddress	read_only_udm.principal.asset.ip	如果 data.ipAddress 欄位不等於 assetId 欄位，則會對應至 principal.asset.ip。
data.ipAddress	read_only_udm.principal.ip	如果 data.ipAddress 欄位不等於 assetId 欄位，則會對應至 principal.ip。
data.location.city	read_only_udm.principal.location.city	如果 location.city 欄位空白，data.location.city 欄位會對應至 principal.location.city。
data.location.countryCode	read_only_udm.principal.location.country_or_region	如果 location.country 欄位為空白，data.location.countryCode 欄位會對應至 principal.location.country_or_region。
data.location.latitude	read_only_udm.principal.location.region_coordinates.latitude	如果 location.coordinates.latitude 和 location.geoCoordinates.latitude 欄位空白，data.location.latitude 欄位會轉換為浮點數，並對應至 principal.location.region_coordinates.latitude。
data.location.longitude	read_only_udm.principal.location.region_coordinates.longitude	如果 location.coordinates.longitude 和 location.geoCoordinates.longitude 欄位為空白，data.location.longitude 欄位會轉換為浮點數，並對應至 principal.location.region_coordinates.longitude。
data.location.province	read_only_udm.principal.location.state	如果 location.province 欄位為空白，data.location.province 欄位會對應至 principal.location.state。
data.mailServers	read_only_udm.additional.fields.value.list_value.values.string_value	data.mailServers 陣列中的每個元素都會對應至個別的 additional.fields 項目，其中鍵為「Mail Servers」，值為 list_value.values.string_value，並設為元素值。
data.names.forwardDns[].name	read_only_udm.network.dns.questions.name	data.names.forwardDns 陣列中的每個元素都會對應至個別的 network.dns.questions 項目，且名稱欄位會設為元素的名稱欄位。
data.nameServers	read_only_udm.additional.fields.value.list_value.values.string_value	data.nameServers 陣列中的每個元素都會對應至個別的 additional.fields 項目，並將索引鍵「Name nameServers」和 value.list_value.values.string_value 設為元素值。
data.protocols[].transportProtocol	read_only_udm.network.ip_protocol	如果 data.protocols[].transportProtocol 欄位是 TCP、EIGRP、ESP、ETHERIP、GRE、ICMP、IGMP、IP6IN4、PIM、UDP 或 VRRP 其中之一，則會對應至 network.ip_protocol。
data.protocols[].transportProtocol	read_only_udm.principal.resource.attribute.labels.value	data.protocols[].transportProtocol 欄位會對應至 principal.resource.attribute.labels.value，且鍵為「data_protocols {index}」。
http.request.headers[].key, http.request.headers[].value.headers.0	read_only_udm.network.http.user_agent	如果 http.request.headers[].key 欄位為「User-Agent」，對應的 http.request.headers[].value.headers.0 欄位會對應至 network.http.user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0	read_only_udm.network.http.parsed_user_agent	如果 http.request.headers[].key 欄位為「User-Agent」，系統會將對應的 http.request.headers[].value.headers.0 欄位剖析為使用者代理程式字串，並對應至 network.http.parsed_user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0	read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value	針對 http.request.headers 陣列中的每個元素，鍵欄位會對應至 principal.resource.attribute.labels.key，而 value.headers.0 欄位會對應至 principal.resource.attribute.labels.value。
http.request.uri	read_only_udm.principal.asset.hostname	系統會擷取 http.request.uri 欄位的主機名稱部分，並對應至 principal.asset.hostname。
http.request.uri	read_only_udm.principal.hostname	系統會擷取 http.request.uri 欄位的主機名稱部分，並對應至 principal.hostname。
http.response.body	read_only_udm.principal.resource.attribute.labels.value	http.response.body 欄位會對應至 principal.resource.attribute.labels.value，且鍵為「http_response_body」。
http.response.headers[].key、http.response.headers[].value.headers.0	read_only_udm.target.hostname	如果 http.response.headers[].key 欄位為「Server」，對應的 http.response.headers[].value.headers.0 欄位會對應至 target.hostname。
http.response.headers[].key、http.response.headers[].value.headers.0	read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value	針對 http.response.headers 陣列中的每個元素，鍵欄位會對應至 principal.resource.attribute.labels.key，而 value.headers.0 欄位會對應至 principal.resource.attribute.labels.value。
http.response.statusCode	read_only_udm.network.http.response_code	http.response.statusCode 欄位會轉換為整數，並對應至 network.http.response_code。
ip	read_only_udm.target.asset.ip	ip 欄位會對應至 target.asset.ip。
ip	read_only_udm.target.ip	ip 欄位會對應至 target.ip。
isSeed	read_only_udm.additional.fields.value.string_value	isSeed 欄位會轉換為字串，並對應至 additional.fields.value.string_value，索引鍵為「isSeed」。
location.city	read_only_udm.principal.location.city	location.city 欄位會對應至 principal.location.city。
location.continent	read_only_udm.additional.fields.value.string_value	location.continent 欄位會對應至 additional.fields.value.string_value，並以「location_continent」做為鍵。
location.coordinates.latitude	read_only_udm.principal.location.region_coordinates.latitude	location.coordinates.latitude 欄位會轉換為浮點數，並對應至 principal.location.region_coordinates.latitude。
location.coordinates.longitude	read_only_udm.principal.location.region_coordinates.longitude	location.coordinates.longitude 欄位會轉換為浮點數，並對應至 principal.location.region_coordinates.longitude。
location.country	read_only_udm.principal.location.country_or_region	location.country 欄位會對應至 principal.location.country_or_region。
location.geoCoordinates.latitude	read_only_udm.principal.location.region_coordinates.latitude	如果 location.coordinates.latitude 欄位為空白，系統會將 location.geoCoordinates.latitude 欄位轉換為浮點數，並對應至 principal.location.region_coordinates.latitude。
location.geoCoordinates.longitude	read_only_udm.principal.location.region_coordinates.longitude	如果 location.coordinates.longitude 欄位空白，系統會將 location.geoCoordinates.longitude 欄位轉換為浮點數，並對應至 principal.location.region_coordinates.longitude。
location.postalCode	read_only_udm.additional.fields.value.string_value	location.postalCode 欄位會對應至 additional.fields.value.string_value，索引鍵為「Postal code」。
location.province	read_only_udm.principal.location.state	location.province 欄位會對應至 principal.location.state。
作業	read_only_udm.security_result.action_details	作業欄位會對應至 security_result.action_details。
perspectiveId	read_only_udm.principal.group.product_object_id	perspectiveId 欄位會對應至 principal.group.product_object_id。
通訊埠	read_only_udm.principal.port	通訊埠欄位會轉換為整數，並對應至 principal.port。
risks[].severity、risks[].title	read_only_udm.security_result.category_details	風險的 []severity 欄位會與風險的 []title 欄位串連，並對應至 security_result.category_details。
serviceName	read_only_udm.network.application_protocol	如果 serviceName 欄位為「HTTP」或「HTTPS」，則會對應至 network.application_protocol。
sourceIp	read_only_udm.principal.asset.ip	sourceIp 欄位會對應至 principal.asset.ip。
sourceIp	read_only_udm.principal.ip	sourceIp 欄位會對應至 principal.ip。
時間戳記	read_only_udm.metadata.event_timestamp	時間戳記欄位會剖析為時間戳記，並對應至 metadata.event_timestamp。
transportFingerprint.id	read_only_udm.metadata.product_log_id	transportFingerprint.id 欄位會轉換為字串，並對應至 metadata.product_log_id。
transportFingerprint.raw	read_only_udm.additional.fields.value.string_value	transportFingerprint.raw 欄位會對應至 additional.fields.value.string_value，並使用「transportFingerprint_raw」鍵。
類型	read_only_udm.metadata.product_event_type	類型欄位會對應至 metadata.product_event_type。
-	read_only_udm.metadata.product_name	值「CENSYS_ASM」會指派給 metadata.product_name。
-	read_only_udm.metadata.vendor_name	值「CENSYS」會指派給 metadata.vendor_name。
-	read_only_udm.metadata.event_type	系統會根據特定欄位是否存在來判斷事件類型：如果 has_princ_machine_id 和 has_target_machine 為 true，且 has_network_flow 為 false，則為 NETWORK_CONNECTION；如果 has_network_flow 為 true，則為 NETWORK_DNS；如果 has_princ_machine_id 為 true，則為 STATUS_UPDATE；否則為 GENERIC_EVENT。

需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。