收集 Cisco 應用程式中心基礎架構 (ACI) 記錄

剖析器版本：4.0

本文說明如何將 Cisco Application Centric Infrastructure (ACI) 記錄檔，擷取至 Google Security Operations。剖析器會先嘗試使用 Grok 模式，將傳入的 Cisco ACI 記錄檔做為 syslog 訊息處理。如果 syslog 剖析失敗，系統會假設訊息採用 JSON 格式，並據此剖析。最後，將擷取的欄位對應至統合式資料模型 (UDM)。

這項整合支援兩種方法：

• 選項 1：使用 Bindplane 代理程式的 Syslog 格式
• 方法 2：使用 APIC REST API 建立 JSON 格式的 Google Cloud Storage

每個選項都是獨立的，您可以根據基礎架構需求和記錄格式偏好，獨立實作這些選項。

選項 1：使用 Bindplane 代理程式的 Syslog

這個選項會設定 Cisco ACI 網路架構，將系統記錄訊息傳送至 Bindplane 代理程式，再由代理程式轉送至 Google Security Operations 進行分析。

事前準備

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

• Google SecOps 執行個體
• 搭載 systemd 的 Windows 2016 以上版本或 Linux 主機
• 如果是在 Proxy 後方執行，請確認防火牆通訊埠已按照 Bindplane 代理程式規定開啟
• Cisco APIC 控制台的特殊權限存取權

取得 Bindplane 代理程式的 Google SecOps 憑證

如要向 Google SecOps 驗證 Bindplane 代理程式，請選擇下列其中一種方法：

• 新部署作業：使用 Chronicle API (選項 1)。
• 舊版部署作業：繼續使用擷取驗證檔案 (選項 2)，直到 Ingestion API 於 2027 年 7 月 20 日停用。

方法 1：為 Chronicle API 建立服務帳戶 (建議)

1. 登入 Google Cloud 控制台，選取連結至 Google SecOps 執行個體的 Google Cloud 專案。
2. 依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
3. 按一下「建立服務帳戶」。
4. 請提供下列設定詳細資料：
   • 服務帳戶名稱：輸入帳戶名稱 (例如 bindplane-ingest)。
   • 服務帳戶說明：輸入說明 (例如 Bindplane agent log ingestion)。
5. 點按「Create and continue」(建立並繼續)。
6. 在「將專案存取權授予這個服務帳戶」部分，按一下「選取角色」，然後選取「Chronicle 編輯者」。
7. 依序點選「繼續」和「完成」。
8. 按一下您建立的服務帳戶，然後前往「金鑰」分頁。
9. 依序點選「新增金鑰」>「建立新的金鑰」。
10. 選取「JSON」，然後點選「建立」。金鑰檔案會下載到電腦。
11. 將金鑰檔案儲存為 Bindplane 收集器目錄中的 ingestion-auth.json (Linux 上的 /opt/observiq-otel-collector/ingestion-auth.json 或 Windows 上的 C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json)，或更新 config.yaml 中的 creds_file_path，使其與儲存路徑相符。
12. 依序前往「IAM & Admin」(IAM 與管理) >「Settings」(設定)，然後複製「Project number」(專案編號)。匯出工具設定時會用到。

方法 2：下載擷取驗證檔案 (舊版)

1. 登入 Google SecOps 控制台。
2. 依序前往「SIEM Settings」 > 「Collection Agents」。
3. 下載擷取驗證檔案。
4. 將檔案安全地儲存在要安裝 Bindplane 代理程式的系統中。

取得 Google SecOps 客戶 ID

1. 依序前往「SIEM 設定」 > 「設定檔」。
2. 複製並儲存「機構詳細資料」部分的客戶 ID。

安裝 Bindplane 代理程式

請按照下列操作說明，在 Windows 或 Linux 作業系統上安裝 Bindplane 代理程式。

Windows 安裝

1. 以管理員身分開啟「命令提示字元」或「PowerShell」。

2. 執行下列指令：

   msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quiet
   

3. 等待安裝完成。

4. 執行下列指令，確認安裝成功：

   sc query observiq-otel-collector
   

   服務應顯示為「RUNNING」。

Linux 安裝

1. 開啟具有根層級或 sudo 權限的終端機。

2. 執行下列指令：

   sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.sh
   

3. 等待安裝完成。

4. 執行下列指令，確認安裝成功：

   sudo systemctl status observiq-otel-collector
   

   服務應顯示為「active (running)」。

其他安裝資源

如需其他安裝選項和疑難排解資訊，請參閱 Bindplane 代理程式安裝指南。

設定 Bindplane 代理程式，擷取系統記錄並傳送至 Google SecOps

存取設定檔

1. 找出 config.yaml 檔案。通常位於 Linux 的 /opt/observiq-otel-collector/ 目錄，或 Windows 的安裝目錄。
2. 使用文字編輯器 (例如 nano、vi 或記事本) 開啟檔案。

編輯 config.yaml 檔案

• config.yaml 檔案

  receivers:
    udplog:
      # Replace the port and IP address as required
      listen_address: "0.0.0.0:514"
  
  exporters:
    chronicle/chronicle_w_labels:
      protocol: https
      # Region of your instance, for example us, europe, asia-southeast1
      location: <LOCATION>
      # Project number of the Google Cloud project linked to the instance
      project: '<PROJECT_NUMBER>'
      compression: gzip
      # Adjust the path to the credentials file you downloaded
      creds_file_path: '/path/to/ingestion-authentication-file.json'
      # Replace with your actual customer ID
      customer_id: <CUSTOMER_ID>
      endpoint: chronicle.googleapis.com
      # Add optional ingestion labels for better organization
      log_type: 'CISCO_ACI'
      raw_log_field: body
      ingestion_labels:
        service:
  
  pipelines:
    logs/source0__chronicle_w_labels-0:
      receivers:
        - udplog
      exporters:
        - chronicle/chronicle_w_labels
  

設定參數

替換下列預留位置：

• listen_address：請根據基礎架構需求，替換通訊埠和 IP 位址。使用 0.0.0.0:514 監聽通訊埠 514 的所有介面。
• creds_file_path：更新為服務帳戶金鑰檔案 (選項 1) 的儲存路徑：
  • Linux：/opt/observiq-otel-collector/ingestion-auth.json
  • Windows：C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
• customer_id：將 YOUR_CUSTOMER_ID 換成上一個步驟中的實際客戶 ID。
• location：將 <LOCATION> 替換為 Google SecOps 執行個體的區域 (例如 us、europe 或 asia-southeast1)。匯出工具會以這個值做為 chronicle.googleapis.com 的前置字元。如需完整清單，請參閱「區域性端點」。
• project：將 <PROJECT_NUMBER> 替換為「IAM 與管理」>「設定」中的專案編號。
• endpoint：維持設為 chronicle.googleapis.com。

舊版 Ingestion API 匯出工具

• 如果您使用方式 2 的擷取驗證檔案，請將 exporters 區塊替換成下列內容。將 creds_file_path 設為擷取驗證檔案的路徑。如果執行個體位於美國境外，請將 malachiteingestion-pa.googleapis.com 替換為區域端點 (例如 europe-malachiteingestion-pa.googleapis.com)。

  exporters:
    chronicle/chronicle_w_labels:
      compression: gzip
      # Adjust the path to the credentials file you downloaded
      creds_file_path: '/path/to/ingestion-authentication-file.json'
      # Replace with your actual customer ID
      customer_id: <CUSTOMER_ID>
      endpoint: malachiteingestion-pa.googleapis.com
      # Add optional ingestion labels for better organization
      log_type: 'CISCO_ACI'
      raw_log_field: body
      ingestion_labels:
        service:
  

儲存設定檔

編輯完畢後，請儲存檔案：

• Linux：依序按下 Ctrl+O、Enter 和 Ctrl+X
• Windows：依序點選「檔案」 > 「儲存」

重新啟動 Bindplane 代理程式，套用變更

Linux

如要在 Linux 系統上重新啟動 Bindplane 代理程式，請按照下列步驟操作：

1. 重新啟動代理：

   sudo systemctl restart observiq-otel-collector
   

2. 確認服務狀態：

   sudo systemctl status observiq-otel-collector
   

3. 追蹤代理記錄，找出錯誤：

   sudo journalctl -u observiq-otel-collector -f
   

Windows

• 如要在 Windows 中重新啟動 Bindplane 代理程式，請選擇下列任一選項：

  • 以管理員身分開啟命令提示字元或 PowerShell：

    net stop observiq-otel-collector && net start observiq-otel-collector
    

  • 服務控制台：

    1. 按下 Win+R 鍵，輸入 services.msc，然後按下 Enter 鍵。
    2. 找出「observIQ OpenTelemetry Collector」。
    3. 按一下滑鼠右鍵，然後選取「重新啟動」。

    4. 確認服務正在執行：

       sc query observiq-otel-collector
       

    5. 查看記錄檔是否有錯誤：

       type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
       

在 Cisco ACI 上設定 Syslog 轉送

設定頻帶外管理合約

1. 登入 Cisco APIC 控制台。
2. 依序前往「Tenants」>「mgmt」>「Contracts」>「Filters」。
3. 按一下「建立篩選器」。
4. 提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 syslog-udp-514。
   • Entry Name：輸入 syslog。
   • 「EtherType」(乙太網路類型)：選取「IP」。
   • IP 通訊協定：選取「UDP」。
   • 「Destination Port Range From」(目的地通訊埠範圍起始值)：輸入 514。
   • 「Destination Port Range To」(目標通訊埠範圍至)：輸入 514。
5. 按一下「提交」。

建立管理合約

1. 依序前往「Tenants」>「mgmt」>「Contracts」>「Standard」。
2. 按一下「建立合約」。
3. 提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 mgmt-syslog-contract。
   • 範圍：選取「Context」。
4. 按一下「提交」。
5. 展開合約，然後按一下「主旨」。
6. 按一下「建立合約主題」。
7. 提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 syslog-subject。
   • 套用兩個方向：勾選這個選項。
8. 按一下「提交」。
9. 展開主題，然後按一下「篩選器」。
10. 按一下「建立篩選器繫結」。
11. 選取「syslog-udp-514」篩選器。
12. 按一下「提交」。

設定 Syslog 目的地群組

1. 依序前往「管理」>「外部資料收集器」>「監控目的地」>「Syslog」。
2. 在「Syslog」上按一下滑鼠右鍵，然後選取「Create Syslog Monitoring Destination Group」。
3. 提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 Chronicle-Syslog-Group。
   • 管理員狀態：選取「已啟用」。
   • 格式：選取「aci」。
4. 點選「下一步」。
5. 在「Create Syslog Monitoring Destination」(建立 Syslog 監控目的地) 對話方塊中：
   • 「Name」(名稱)：輸入 Chronicle-BindPlane。
   • 主機：輸入 Bindplane 代理程式伺服器的 IP 位址。
   • 通訊埠：輸入 514。
   • 管理狀態：選取「已啟用」。
   • 嚴重程度：選取「資訊」 (擷取詳細記錄)。
6. 按一下「提交」。

設定監控政策 - Fabric 監控

1. 依序前往 「Fabric」 > 「Fabric Policies」 > 「Policies」 > 「Monitoring」 > 「Common Policy」。
2. 展開「Callhome/Smart Callhome/SNMP/Syslog/TACACS」。
3. 在「Syslog」上按一下滑鼠右鍵，然後選取「Create Syslog Source」(建立 Syslog 來源)。
4. 提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 Chronicle-Fabric-Syslog。
   • 稽核記錄：勾選即可納入稽核事件。
   • 事件：勾選即可納入系統事件。
   • Faults：勾選即可納入錯誤事件。
   • 工作階段記錄：勾選即可納入工作階段記錄。
   • 目的地群組：選取「Chronicle-Syslog-Group」。
5. 按一下「提交」。

設定監控政策 - 存取監控

1. 依序前往「Fabric」>「Access Policies」>「Policies」>「Monitoring」>「Default Policy」。
2. 展開「Callhome/Smart Callhome/SNMP/Syslog」。
3. 在「Syslog」上按一下滑鼠右鍵，然後選取「Create Syslog Source」。
4. 請提供下列設定詳細資料：
   • 「Name」(名稱)：輸入 Chronicle-Access-Syslog。
   • 稽核記錄：勾選即可納入稽核事件。
   • 事件：勾選即可納入系統事件。
   • Faults：勾選即可納入錯誤事件。
   • 工作階段記錄：勾選即可納入工作階段記錄。
   • 目的地群組：選取「Chronicle-Syslog-Group」。
5. 按一下「提交」。

設定系統 Syslog 訊息政策

1. 依序前往「Fabric」>「Fabric Policies」>「Policies」>「Monitoring」>「Common Policy」。
2. 展開「Syslog Messages Policies」。
3. 點選「default」。
4. 在「Facility Filter」(設施篩選器) 部分：
   • Facility：選取「default」。
   • 「Minimum Severity」(最低嚴重性)：變更為「information」(資訊)。
5. 按一下「提交」。

選項 2：使用 Google Cloud Storage 的 JSON

這個選項會使用 APIC REST API，從 Cisco ACI 結構收集 JSON 格式的事件、錯誤和稽核記錄，並儲存在 Google Cloud Storage 中，供 Google SecOps 擷取。

事前準備

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

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

收集 Cisco ACI APIC 先決條件

取得 APIC 憑證

1. 使用 HTTPS 登入 Cisco APIC 控制台。

2. 依序前往「Admin」>「AAA」 (APIC 6.0 以上版本)，或「Admin」>「Authentication」>「AAA」 (舊版)。

3. 建立或使用具備適當權限的現有本機使用者。

4. 複製下列詳細資料並儲存在安全位置：

   • APIC 使用者名稱：具有監控資料讀取權的本機使用者
   • APIC 密碼：使用者密碼
   • APIC 網址：APIC 的 HTTPS 網址 (例如 https://apic.example.com)

建立 Google Cloud Storage bucket

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

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

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

6. 點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要服務帳戶，並具備寫入 GCS bucket 的權限，且可由 Pub/Sub 叫用。

建立服務帳戶

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

這些角色適用於：

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

授予 GCS bucket 的 IAM 權限

將 GCS bucket 的寫入權限授予服務帳戶 (cisco-aci-collector-sa)：

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

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。

1. 在 GCP 控制台中，依序前往「Pub/Sub」>「Topics」(主題)。
2. 按一下「建立主題」。
3. 提供下列設定詳細資料：
   • 主題 ID：輸入 cisco-aci-trigger。
   • 其他設定保留預設值。
4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

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

1. 前往 GCP 控制台的「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」 (使用直接修改編輯器建立函式)。

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

   設定	值
   服務名稱	cisco-aci-collector
   區域	選取與 GCS bucket 相符的區域 (例如 us-central1)
   執行階段	選取「Python 3.12」或更新版本

5. 在「觸發條件 (選用)」部分：

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

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

   1. 選取「需要驗證」。
   2. Identity and Access Management (IAM)。

7. 向下捲動並展開「容器、網路、安全性」。

8. 前往「安全性」分頁：

   • 服務帳戶：選取服務帳戶 (cisco-aci-collector-sa)。

9. 前往「容器」分頁：

   1. 按一下「Variables & Secrets」(變數與密鑰)。
   2. 針對每個環境變數，按一下「+ 新增變數」：

   變數名稱	範例值	說明
   GCS_BUCKET	cisco-aci-logs	GCS bucket 名稱
   GCS_PREFIX	cisco-aci-events	記錄檔的前置字串
   STATE_KEY	cisco-aci-events/state.json	狀態檔案路徑
   APIC_URL	https://apic.example.com	APIC HTTPS 網址
   APIC_USERNAME	your-apic-username	APIC 使用者名稱
   APIC_PASSWORD	your-apic-password	APIC 密碼
   PAGE_SIZE	100	每頁記錄數
   MAX_PAGES	10	每次執行作業的頁數上限

10. 在「Variables & Secrets」(變數與 Secret) 部分，捲動至「Requests」(要求)：

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

11. 前往「設定」分頁：

    • 在「Resources」(資源) 部分：
      • 記憶體：選取 512 MiB 以上。
      • CPU：選取 1。

12. 在「Revision scaling」(修訂版本資源調度) 部分：

    • 執行個體數量下限：輸入 0。
    • 「執行個體數量上限」：輸入 100 (或依預期負載調整)。

13. 點選「建立」。

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

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

新增函式程式碼

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 logging
   
   # Configure logging
   logger = logging.getLogger()
   logger.setLevel(logging.INFO)
   
   # Initialize HTTP client with timeouts
   http = urllib3.PoolManager(
       timeout=urllib3.Timeout(connect=5.0, read=60.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', 'cisco-aci-events')
   STATE_KEY = os.environ.get('STATE_KEY', 'cisco-aci-events/state.json')
   APIC_URL = os.environ.get('APIC_URL')
   APIC_USERNAME = os.environ.get('APIC_USERNAME')
   APIC_PASSWORD = os.environ.get('APIC_PASSWORD')
   PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
   MAX_PAGES = int(os.environ.get('MAX_PAGES', '10'))
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch Cisco ACI logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       if not all([GCS_BUCKET, APIC_URL, APIC_USERNAME, APIC_PASSWORD]):
           logger.error('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
           last_timestamp = state.get('last_timestamp')
           if not last_timestamp:
               last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z'
   
           logger.info(f"Starting Cisco ACI data collection for bucket: {GCS_BUCKET}")
   
           # Authenticate to APIC
           session_token = authenticate_apic(APIC_URL, APIC_USERNAME, APIC_PASSWORD)
           headers = {
               'Cookie': f'APIC-cookie={session_token}',
               'Accept': 'application/json',
               'Content-Type': 'application/json'
           }
   
           # Data types to collect
           data_types = ['faultInst', 'eventRecord', 'aaaModLR']
           all_collected_data = []
   
           for data_type in data_types:
               logger.info(f"Collecting {data_type} data")
               collected_data = collect_aci_data(
                   APIC_URL,
                   headers,
                   data_type,
                   last_timestamp,
                   PAGE_SIZE,
                   MAX_PAGES
               )
   
               # Tag each record with its type
               for record in collected_data:
                   record['_data_type'] = data_type
   
               all_collected_data.extend(collected_data)
               logger.info(f"Collected {len(collected_data)} {data_type} records")
   
           logger.info(f"Total records collected: {len(all_collected_data)}")
   
           # Store data in GCS if any were collected
           if all_collected_data:
               timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
               s3_key = f"{GCS_PREFIX}/cisco_aci_events_{timestamp_str}.ndjson"
   
               # Convert to NDJSON format (one JSON object per line)
               ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data)
   
               # Upload to GCS
               blob = bucket.blob(s3_key)
               blob.upload_from_string(
                   ndjson_content,
                   content_type='application/x-ndjson'
               )
   
               logger.info(f"Uploaded {len(all_collected_data)} records to gs://{GCS_BUCKET}/{s3_key}")
   
               # Update state file with latest timestamp from collected data
               latest_timestamp = get_latest_timestamp_from_records(all_collected_data)
               if not latest_timestamp:
                   latest_timestamp = datetime.utcnow().isoformat() + 'Z'
   
               update_state(bucket, STATE_KEY, latest_timestamp)
           else:
               logger.info("No new log records found.")
   
           logger.info(f"Successfully processed {len(all_collected_data)} records")
   
       except Exception as e:
           logger.error(f'Error processing logs: {str(e)}')
           raise
   
   def authenticate_apic(apic_url, username, password):
       """Authenticate to APIC and return session token"""
       login_url = f"{apic_url}/api/aaaLogin.json"
       login_data = {
           "aaaUser": {
               "attributes": {
                   "name": username,
                   "pwd": password
               }
           }
       }
   
       response = http.request(
           'POST',
           login_url,
           body=json.dumps(login_data).encode('utf-8'),
           headers={'Content-Type': 'application/json'},
           timeout=30
       )
   
       if response.status != 200:
           raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}")
   
       response_data = json.loads(response.data.decode('utf-8'))
       token = response_data['imdata'][0]['aaaLogin']['attributes']['token']
       logger.info("Successfully authenticated to APIC")
       return token
   
   def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages):
       """Collect data from APIC REST API with pagination"""
       all_data = []
       page = 0
   
       while page < max_pages:
           # Build API URL with pagination and time filters
           api_url = f"{apic_url}/api/class/{data_type}.json"
           params = [
               f'page-size={page_size}',
               f'page={page}',
               f'order-by={data_type}.created|asc'
           ]
   
           # Add time filter to prevent duplicates
           if last_timestamp:
               params.append(f'query-target-filter=gt({data_type}.created,"{last_timestamp}")')
   
           full_url = f"{api_url}?{'&'.join(params)}"
   
           logger.info(f"Fetching {data_type} page {page} from APIC")
   
           # Make API request
           response = http.request('GET', full_url, headers=headers, timeout=60)
   
           if response.status != 200:
               logger.error(f"API request failed: {response.status} {response.data[:256]!r}")
               break
   
           data = json.loads(response.data.decode('utf-8'))
           records = data.get('imdata', [])
   
           if not records:
               logger.info(f"No more {data_type} records found")
               break
   
           # Extract the actual data from APIC format
           extracted_records = []
           for record in records:
               if data_type in record:
                   extracted_records.append(record[data_type])
   
           all_data.extend(extracted_records)
           page += 1
   
           # If we got less than page_size records, we've reached the end
           if len(records) < page_size:
               break
   
       return all_data
   
   def get_last_timestamp(bucket, state_key):
       """Get the last run timestamp from GCS state file"""
       try:
           blob = bucket.blob(state_key)
           if blob.exists():
               state_data = blob.download_as_text()
               state = json.loads(state_data)
               return state.get('last_timestamp')
       except Exception as e:
           logger.warning(f"Error reading state file: {str(e)}")
   
       return None
   
   def get_latest_timestamp_from_records(records):
       """Get the latest timestamp from collected records to prevent missing events"""
       if not records:
           return None
   
       latest = None
       latest_time = None
   
       for record in records:
           try:
               # Handle both direct attributes and nested structure
               attrs = record.get('attributes', record)
               created = attrs.get('created')
               modTs = attrs.get('modTs')
   
               # Use created or modTs as fallback
               timestamp = created or modTs
   
               if timestamp:
                   if latest_time is None or timestamp > latest_time:
                       latest_time = timestamp
                       latest = record
           except Exception as e:
               logger.debug(f"Error parsing timestamp from record: {e}")
               continue
   
       return latest_time
   
   def update_state(bucket, state_key, timestamp):
       """Update the state file with the current timestamp"""
       try:
           state_data = {
               'last_timestamp': timestamp,
               'updated_at': datetime.utcnow().isoformat() + 'Z'
           }
           blob = bucket.blob(state_key)
           blob.upload_from_string(
               json.dumps(state_data),
               content_type='application/json'
           )
           logger.info(f"Updated state file with timestamp: {timestamp}")
       except Exception as e:
           logger.error(f"Error updating state file: {str(e)}")
   
   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:
           logger.warning(f"Could not load state: {e}")
   
       return {}
   

   • 第二個檔案：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. 提供下列設定詳細資料：

   設定	值
   名稱	cisco-aci-collector-15m
   區域	選取與 Cloud Run 函式相同的區域
   頻率	*/15 * * * * (每 15 分鐘)
   時區	選取時區 (建議選取世界標準時間)
   目標類型	Pub/Sub
   主題	選取 Pub/Sub 主題 (cisco-aci-trigger)
   郵件內文	{} (空白 JSON 物件)

4. 點選「建立」。

排程頻率選項

請根據記錄量和延遲時間需求選擇頻率：

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

測試整合

1. 在 Cloud Scheduler 控制台中，找到您的工作 (cisco-aci-collector-15m)。
2. 按一下「強制執行」，手動觸發工作。
3. 稍等幾秒鐘。
4. 依序前往「Cloud Run」 > 「服務」。
5. 按一下下列函式名稱 (cisco-aci-collector)。
6. 點選「記錄」分頁標籤。

7. 確認函式是否執行成功。尋找：

   Starting Cisco ACI data collection for bucket: cisco-aci-logs
   Successfully authenticated to APIC
   Collecting faultInst data
   Collected X faultInst records
   Collecting eventRecord data
   Collected X eventRecord records
   Collecting aaaModLR data
   Collected X aaaModLR records
   Total records collected: X
   Uploaded X records to gs://cisco-aci-logs/cisco-aci-events/cisco_aci_events_YYYYMMDD_HHMMSS.ndjson
   Successfully processed X records
   

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

9. 按一下 bucket 名稱 (cisco-aci-logs)。

10. 前往前置字元資料夾 (cisco-aci-events/)。

11. 確認系統已建立新的 .ndjson 檔案，並加上目前的時間戳記。

如果記錄中顯示錯誤：

• HTTP 401：檢查環境變數中的 APIC 憑證
• HTTP 403：確認 APIC 帳戶是否具有 faultInst、eventRecord 和 aaaModLR 類別的讀取權限
• 連線錯誤：確認 Cloud Run 函式可透過 TCP/443 連線至 APIC 網址
• 缺少環境變數：確認已設定所有必要變數

在 Google SecOps 設定動態饋給，擷取 Cisco ACI 記錄

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態饋給」。
3. 按一下「設定單一動態饋給」。
4. 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如 Cisco ACI JSON logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「Cisco Application Centric Infrastructure」做為「記錄類型」。

7. 按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：

   chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
   

8. 複製這個電子郵件地址。後續步驟會用到。

9. 點選「下一步」。

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

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

    gs://cisco-aci-logs/cisco-aci-events/
    

    • 取代：
      • cisco-aci-logs：您的 GCS bucket 名稱。
      • cisco-aci-events：：儲存記錄的前置字元/資料夾路徑。
    • 來源刪除選項：根據偏好選取刪除選項：
      • Never：轉移後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空目錄：成功轉移後，刪除檔案和空目錄。
    • 檔案存在時間上限：包含過去幾天內修改過的檔案。預設值為 180 天。
    • Asset namespace：資產命名空間。
    • 擷取標籤：套用至這個動態饋給事件的標籤。

11. 點選「下一步」。

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

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

Google SecOps 服務帳戶必須具備 GCS bucket 的 Storage 物件檢視者角色。

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

6. 按一下 [儲存]。

UDM 對應表

記錄欄位	UDM 對應	邏輯
@timestamp	read_only_udm.metadata.event_timestamp	值取自原始記錄欄位「@timestamp」，並剖析為時間戳記。
aci_tag	read_only_udm.metadata.product_log_id	值取自原始記錄欄位「aci_tag」。
cisco_timestamp	-	未對應。
DIP	read_only_udm.target.ip	值取自原始記錄欄位「DIP」。
DPort	read_only_udm.target.port	值取自原始記錄欄位「DPort」，並轉換為整數。
說明	read_only_udm.security_result.description	值取自原始記錄欄位「description」。
fault_cause	read_only_udm.additional.fields.value.string_value	值取自原始記錄欄位「fault_cause」。鍵設為「Fault Cause」。
主機名稱	read_only_udm.principal.hostname	值取自原始記錄欄位「hostname」。
lifecycle_state	read_only_udm.metadata.product_event_type	值取自原始記錄欄位「lifecycle_state」。
log.source.address	-	未對應。
logstash.collect.host	-	未對應。
logstash.collect.timestamp	read_only_udm.metadata.collected_timestamp	值取自原始記錄欄位「logstash.collect.timestamp」，並剖析為時間戳記。
logstash.ingest.host	read_only_udm.intermediary.hostname	值取自原始記錄欄位「logstash.ingest.host」。
logstash.irm_environment	read_only_udm.additional.fields.value.string_value	值取自原始記錄檔欄位「logstash.irm_environment」。金鑰設為「IRM_Environment」。
logstash.irm_region	read_only_udm.additional.fields.value.string_value	值取自原始記錄欄位「logstash.irm_region」。金鑰設為「IRM_Region」。
logstash.irm_site	read_only_udm.additional.fields.value.string_value	值取自原始記錄欄位「logstash.irm_site」。鍵設為「IRM_Site」。
logstash.process.host	read_only_udm.intermediary.hostname	值取自原始記錄欄位「logstash.process.host」。
訊息	-	未對應。
message_class	-	未對應。
message_code	-	未對應。
message_content	-	未對應。
message_dn	-	未對應。
message_type	read_only_udm.metadata.product_event_type	這個值取自原始記錄欄位「message_type」(移除方括號後)。
node_link	read_only_udm.principal.process.file.full_path	值取自原始記錄欄位「node_link」。
PktLen	read_only_udm.network.received_bytes	值取自原始記錄欄位「PktLen」，並轉換為不帶正負號的整數。
計畫	-	未對應。
Proto	read_only_udm.network.ip_protocol	值取自原始記錄欄位「Proto」，轉換為整數，並對應至相應的 IP 通訊協定名稱 (例如 6 -> TCP)。
SIP	read_only_udm.principal.ip	值取自原始記錄欄位「SIP」。
SPort	read_only_udm.principal.port	值取自原始記錄欄位「SPort」，並轉換為整數。
syslog_facility	-	未對應。
syslog_facility_code	-	未對應。
syslog_host	read_only_udm.principal.ip、read_only_udm.observer.ip	值取自原始記錄欄位「syslog_host」。
syslog_prog	-	未對應。
syslog_severity	read_only_udm.security_result.severity_details	值取自原始記錄檔欄位「syslog_severity」。
syslog_severity_code	read_only_udm.security_result.severity	這個值取自原始記錄欄位「syslog_severity_code」，並對應至相應的嚴重性等級：5、6、7 -> INFORMATIONAL；3、4 -> MEDIUM；0、1、2 -> HIGH。
syslog5424_pri	-	未對應。
Vlan-Id	read_only_udm.principal.resource.id	值取自原始記錄欄位「Vlan-Id」。
-	read_only_udm.metadata.event_type	邏輯：如果「SIP」或「hostname」和「Proto」存在，則設為「NETWORK_CONNECTION」。否則，如果存在「SIP」、「hostname」或「syslog_host」，則設為「STATUS_UPDATE」。否則，請設為「GENERIC_EVENT」。
-	read_only_udm.metadata.log_type	邏輯：設為「CISCO_ACI」。
-	read_only_udm.metadata.vendor_name	邏輯：設為「Cisco」。
-	read_only_udm.metadata.product_name	邏輯：設為「ACI」。
principal_mac	event.idm.read_only_udm.principal.mac 和 event.idm.read_only_udm.principal.asset.mac	從變更記錄對應
context_id	event.idm.read_only_udm.additional.fields	從變更記錄對應
node_number	event.idm.read_only_udm.additional.fields	從變更記錄對應
code1	event.idm.read_only_udm.security_result.detection_fields	從變更記錄對應
code2	event.idm.read_only_udm.security_result.detection_fields	從變更記錄對應
pid	event.idm.read_only_udm.principal.process.pid	從變更記錄對應
src_intf	event.idm.read_only_udm.principal.resource.attribute.labels	從變更記錄對應
VXLAN	event.idm.read_only_udm.network.session_id	從變更記錄對應
VlanType	event.idm.read_only_udm.additional.fields	從變更記錄對應
flag	event.idm.read_only_udm.additional.fields	從變更記錄對應
Hitcnt	event.idm.read_only_udm.additional.fields	從變更記錄對應
syslog_facility	event.idm.read_only_udm.additional.fields	從變更記錄對應
SMac	event.idm.read_only_udm.principal.resource.attribute.labels	從變更記錄對應
DMac	event.idm.read_only_udm.target.resource.attribute.labels	從變更記錄對應
path	event.idm.read_only_udm.principal.process.file.full_path	從變更記錄對應
CName	event.idm.read_only_udm.security_result.rule_name	從變更記錄對應
i_ip	event.idm.read_only_udm.intermediary.ip	從變更記錄對應
i_ip	event.idm.read_only_udm.intermediary.asset.ip	從變更記錄對應
log_id	event.idm.read_only_udm.metadata.product_log_id	從變更記錄對應
product_event_type	event.idm.read_only_udm.metadata.product_event_type	從變更記錄對應
severity_action	event.idm.read_only_udm.security_result.severity	從變更記錄對應
session_id	event.idm.read_only_udm.network.session_id	從變更記錄對應
protocol	event.idm.read_only_udm.additional.fields	從變更記錄對應
message_code	event.idm.read_only_udm.additional.fields	從變更記錄對應
program	event.idm.read_only_udm.additional.fields	從變更記錄對應
syslog_prog	event.idm.read_only_udm.additional.fields	從變更記錄對應
message_dn	event.idm.read_only_udm.additional.fields	從變更記錄對應
message_class	event.idm.read_only_udm.additional.fields	從變更記錄對應
syslog_facility_code	event.idm.read_only_udm.additional.fields	從變更記錄對應
syslog5424_pri	event.idm.read_only_udm.additional.fields	從變更記錄對應
initial_code	event.idm.read_only_udm.additional.fields	從變更記錄對應
logstash.ingest.timestamp	event.idm.read_only_udm.additional.fields	從變更記錄對應
cisco_timestamp	event.idm.read_only_udm.additional.fields	從變更記錄對應

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。