收集 Cisco vManage SD-WAN 記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，將 Cisco vManage SD-WAN 記錄檔擷取至 Google Security Operations。Cisco vManage SD-WAN 是集中式網路管理系統，可提供 SD-WAN 結構的可視性和控制權，讓管理員監控網路效能、設定政策，以及管理分散式企業網路的安全。

事前準備

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

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
Cisco vManage SD-WAN 管理控制台的特殊存取權
具有 API 存取權限的 Cisco vManage 使用者帳戶

建立 Google Cloud Storage 值區

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

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

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

點選「建立」。

收集 Cisco vManage SD-WAN API 憑證

建立 API 使用者帳戶

登入 Cisco vManage 管理控制台。
依序前往「管理」>「設定」>「使用者」。

按一下「新增使用者」。

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

使用者名稱：輸入 API 存取權的使用者名稱 (例如 chronicle-api)。

密碼：輸入高強度密碼。

確認密碼：再次輸入密碼。

使用者群組：選取具備適當權限的使用者群組 (請參閱下一節)。

按一下 [新增]。

複製下列詳細資料並存放在安全的地方：

使用者名稱：您的 vManage 使用者名稱。

密碼：您的 vManage 密碼。

vManage 基準網址：vManage 伺服器的基準網址 (例如 https://your-vmanage-server:8443)。

注意： Cloud Run 函式會使用這些憑證向 Cisco vManage API 進行驗證。

設定使用者權限

API 使用者帳戶必須具備特定權限，才能存取稽核記錄、警報和事件。

在 Cisco vManage 管理控制台中，依序前往「Administration」>「Settings」>「User Groups」。

選取指派給 API 使用者的使用者群組 (或建立新群組)。

按一下 [編輯]。

在「功能」部分，確認已啟用下列權限：

稽核記錄：選取「讀取」權限。

鬧鐘：選取「讀取」權限。

事件：選取「讀取」權限。

按一下「更新」。

驗證 API 存取權

請先測試憑證，再繼續進行整合：

開啟終端機或命令提示字元。

執行下列指令來測試驗證：

# Replace with your actual credentials VMANAGE_HOST="https://your-vmanage-server:8443" VMANAGE_USERNAME="chronicle-api" VMANAGE_PASSWORD="your-password" # Test authentication (returns JSESSIONID cookie) curl -c cookies.txt -X POST \ "${VMANAGE_HOST}/j_security_check" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "j_username=${VMANAGE_USERNAME}&j_password=${VMANAGE_PASSWORD}" # Get CSRF token curl -b cookies.txt \ "${VMANAGE_HOST}/dataservice/client/token"

如果驗證成功，第二個指令會傳回 CSRF 權杖字串。

Note: In production environments, configure valid TLS certificates on vManage and verify certificates in the HTTP client. The code examples use certificate verification disabled for testing purposes only.

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。

按一下 [Create Service Account] (建立服務帳戶)。

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

服務帳戶名稱：輸入 cisco-sdwan-collector-sa。

服務帳戶說明：輸入 Service account for Cloud Run function to collect Cisco vManage SD-WAN logs。

按一下「建立並繼續」。

在「將專案存取權授予這個服務帳戶」部分，新增下列角色：

按一下「選擇角色」。

搜尋並選取「Storage 物件管理員」。

點選「+ 新增其他角色」。

搜尋並選取「Cloud Run Invoker」。

點選「+ 新增其他角色」。

搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。

按一下「繼續」。

按一下 [完成]。

這些角色適用於：

Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案

Cloud Run 叫用者：允許 Pub/Sub 叫用函式

Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

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

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

按一下 bucket 名稱。

前往「權限」分頁標籤。

按一下「授予存取權」。

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

新增主體：輸入服務帳戶電子郵件地址 (例如 cisco-sdwan-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。

指派角色：選取「Storage 物件管理員」。

按一下 [儲存]。

建立 Pub/Sub 主題

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

在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。

按一下「建立主題」。

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

主題 ID：輸入 cisco-sdwan-trigger。

其他設定保留預設值。

點選「建立」。

建立 Cloud Run 函式來收集記錄

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

前往 GCP Console 的「Cloud Run」。

按一下「Create service」(建立服務)。

選取「函式」 (使用內嵌編輯器建立函式)。

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

設定值

服務名稱 cisco-sdwan-log-collector

區域選取與 GCS bucket 相符的區域 (例如 us-central1)

執行階段 選取「Python 3.12」以上版本

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

按一下「+ 新增觸發條件」。

選取「Cloud Pub/Sub」。

在「選取 Cloud Pub/Sub 主題」中，選擇主題 (cisco-sdwan-trigger)。

按一下 [儲存]。

在「Authentication」(驗證) 部分：

選取「需要驗證」。

檢查 Identity and Access Management (IAM)。

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

向下捲動並展開「Containers, Networking, Security」。

前往「安全性」分頁：

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

前往「容器」分頁：

按一下「變數與密鑰」。

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

變數名稱範例值

GCS_BUCKET cisco-sdwan-logs-bucket

GCS_PREFIX cisco-sdwan/

STATE_KEY cisco-sdwan/state.json

VMANAGE_HOST https://your-vmanage-server:8443

VMANAGE_USERNAME chronicle-api

VMANAGE_PASSWORD your-vmanage-password

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

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

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

在「資源」部分：

記憶體：選取 512 MiB 以上。

CPU：選取 1。

按一下 [完成]。

向下捲動至「執行環境」：

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

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

執行個體數量下限：輸入 0。

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

點選「建立」。

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

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

新增函式程式碼

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

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

第一個檔案：main.py:

import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone import time # Disable SSL warnings for self-signed certificates (testing only) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.0), cert_reqs='ssl.CERT_NONE', retries=urllib3.Retry(total=3, backoff_factor=1) ) # Environment variables VMANAGE_HOST = os.environ['VMANAGE_HOST'] VMANAGE_USERNAME = os.environ['VMANAGE_USERNAME'] VMANAGE_PASSWORD = os.environ['VMANAGE_PASSWORD'] GCS_BUCKET = os.environ['GCS_BUCKET'] GCS_PREFIX = os.environ['GCS_PREFIX'] STATE_KEY = os.environ['STATE_KEY'] # Initialize clients storage_client = storage.Client() class VManageAPI: def __init__(self, host, username, password): self.host = host.rstrip('/') self.username = username self.password = password self.cookies = None self.token = None def authenticate(self): """Authenticate with vManage and get session tokens""" try: # Login to get JSESSIONID login_url = f"{self.host}/j_security_check" # Encode credentials properly import urllib.parse login_data = urllib.parse.urlencode({ 'j_username': self.username, 'j_password': self.password }).encode('utf-8') response = http.request( 'POST', login_url, body=login_data, headers={'Content-Type': 'application/x-www-form-urlencoded'}, ) # Check if login was successful if b'<html>' in response.data or response.status != 200: print(f"Authentication failed: HTTP {response.status}") return False # Extract cookies self.cookies = {} if 'Set-Cookie' in response.headers: cookie_header = response.headers['Set-Cookie'] for cookie in cookie_header.split(';'): if 'JSESSIONID=' in cookie: self.cookies['JSESSIONID'] = cookie.split('JSESSIONID=')[1].split(';')[0] break if not self.cookies.get('JSESSIONID'): print("Failed to get JSESSIONID") return False # Get XSRF token token_url = f"{self.host}/dataservice/client/token" headers = { 'Content-Type': 'application/json', 'Cookie': f"JSESSIONID={self.cookies['JSESSIONID']}" } response = http.request('GET', token_url, headers=headers) if response.status == 200: self.token = response.data.decode('utf-8') print("Successfully authenticated with vManage") return True else: print(f"Failed to get XSRF token: HTTP {response.status}") return False except Exception as e: print(f"Authentication error: {e}") return False def get_headers(self): """Get headers for API requests""" return { 'Content-Type': 'application/json', 'Cookie': f"JSESSIONID={self.cookies['JSESSIONID']}", 'X-XSRF-TOKEN': self.token } def get_audit_logs(self, last_timestamp=None): """Get audit logs from vManage""" try: url = f"{self.host}/dataservice/auditlog" headers = self.get_headers() query = { "query": { "condition": "AND", "rules": [] }, "size": 10000 } if last_timestamp: if isinstance(last_timestamp, str): try: dt = datetime.fromisoformat(last_timestamp.replace('Z', '+00:00')) epoch_ms = int(dt.timestamp() * 1000) except: epoch_ms = int(last_timestamp) else: epoch_ms = int(last_timestamp) query["query"]["rules"].append({ "value": [str(epoch_ms)], "field": "entry_time", "type": "date", "operator": "greater" }) else: query["query"]["rules"].append({ "value": ["1"], "field": "entry_time", "type": "date", "operator": "last_n_hours" }) response = http.request( 'POST', url, body=json.dumps(query), headers=headers, ) if response.status == 200: return json.loads(response.data.decode('utf-8')) else: print(f"Failed to get audit logs: HTTP {response.status}") return None except Exception as e: print(f"Error getting audit logs: {e}") return None def get_alarms(self, last_timestamp=None): """Get alarms from vManage""" try: url = f"{self.host}/dataservice/alarms" headers = self.get_headers() query = { "query": { "condition": "AND", "rules": [] }, "size": 10000 } if last_timestamp: if isinstance(last_timestamp, str): try: dt = datetime.fromisoformat(last_timestamp.replace('Z', '+00:00')) epoch_ms = int(dt.timestamp() * 1000) except: epoch_ms = int(last_timestamp) else: epoch_ms = int(last_timestamp) query["query"]["rules"].append({ "value": [str(epoch_ms)], "field": "entry_time", "type": "date", "operator": "greater" }) else: query["query"]["rules"].append({ "value": ["1"], "field": "entry_time", "type": "date", "operator": "last_n_hours" }) response = http.request( 'POST', url, body=json.dumps(query), headers=headers, ) if response.status == 200: return json.loads(response.data.decode('utf-8')) else: print(f"Failed to get alarms: HTTP {response.status}") return None except Exception as e: print(f"Error getting alarms: {e}") return None def get_events(self, last_timestamp=None): """Get events from vManage""" try: url = f"{self.host}/dataservice/events" headers = self.get_headers() query = { "query": { "condition": "AND", "rules": [] }, "size": 10000 } if last_timestamp: if isinstance(last_timestamp, str): try: dt = datetime.fromisoformat(last_timestamp.replace('Z', '+00:00')) epoch_ms = int(dt.timestamp() * 1000) except: epoch_ms = int(last_timestamp) else: epoch_ms = int(last_timestamp) query["query"]["rules"].append({ "value": [str(epoch_ms)], "field": "entry_time", "type": "date", "operator": "greater" }) else: query["query"]["rules"].append({ "value": ["1"], "field": "entry_time", "type": "date", "operator": "last_n_hours" }) response = http.request( 'POST', url, body=json.dumps(query), headers=headers, ) if response.status == 200: return json.loads(response.data.decode('utf-8')) else: print(f"Failed to get events: HTTP {response.status}") return None except Exception as e: print(f"Error getting events: {e}") return None def get_last_run_time(bucket): """Get the last successful run timestamp from GCS""" try: blob = bucket.blob(STATE_KEY) if blob.exists(): state_data = json.loads(blob.download_as_text()) return state_data.get('last_run_time') except Exception as e: print(f"Error reading state: {e}") print("No previous state found, collecting last hour of logs") return None def update_last_run_time(bucket, timestamp): """Update the last successful run timestamp in GCS""" try: state_data = { 'last_run_time': timestamp, 'updated_at': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(STATE_KEY) blob.upload_from_string( json.dumps(state_data), content_type='application/json' ) print(f"Updated state with timestamp: {timestamp}") except Exception as e: print(f"Error updating state: {e}") def upload_logs_to_gcs(bucket, logs_data, log_type, timestamp): """Upload logs to GCS bucket""" try: if not logs_data or 'data' not in logs_data or not logs_data['data']: print(f"No {log_type} data to upload") return dt = datetime.now(timezone.utc) filename = f"{GCS_PREFIX}{log_type}/{dt.strftime('%Y/%m/%d')}/{log_type}_{dt.strftime('%Y%m%d_%H%M%S')}.json" blob = bucket.blob(filename) blob.upload_from_string( json.dumps(logs_data), content_type='application/json' ) print(f"Uploaded {len(logs_data['data'])} {log_type} records to gs://{GCS_BUCKET}/{filename}") except Exception as e: print(f"Error uploading {log_type} to GCS: {e}") @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from Cisco vManage API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ print(f"Starting Cisco vManage log collection at {datetime.now(timezone.utc)}") try: bucket = storage_client.bucket(GCS_BUCKET) # Get last run time last_run_time = get_last_run_time(bucket) # Initialize vManage API client vmanage = VManageAPI(VMANAGE_HOST, VMANAGE_USERNAME, VMANAGE_PASSWORD) # Authenticate if not vmanage.authenticate(): print('Failed to authenticate with vManage') return # Current timestamp for state tracking (store as epoch milliseconds) current_time = int(datetime.now(timezone.utc).timestamp() * 1000) # Collect different types of logs log_types = [ ('audit_logs', vmanage.get_audit_logs), ('alarms', vmanage.get_alarms), ('events', vmanage.get_events) ] total_records = 0 for log_type, get_function in log_types: try: print(f"Collecting {log_type}...") logs_data = get_function(last_run_time) if logs_data: upload_logs_to_gcs(bucket, logs_data, log_type, current_time) if 'data' in logs_data: total_records += len(logs_data['data']) except Exception as e: print(f"Error processing {log_type}: {e}") continue # Update state with current timestamp update_last_run_time(bucket, current_time) print(f"Collection completed. Total records processed: {total_records}") except Exception as e: print(f"Function execution error: {e}") raise

第二個檔案：requirements.txt:

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

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

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

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

建立 Cloud Scheduler 工作

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

前往 GCP 主控台的「Cloud Scheduler」。

點選「建立工作」。

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

設定值

名稱 cisco-sdwan-log-collector-hourly

區域選取與 Cloud Run 函式相同的區域

頻率 0 * * * * (每小時整點)

時區選取時區 (建議使用世界標準時間)

目標類型 Pub/Sub

主題選取主題 (cisco-sdwan-trigger)

郵件內文 {} (空白 JSON 物件)

點選「建立」。

排程頻率選項

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

頻率 Cron 運算式用途

每 5 分鐘 */5 * * * * 高容量、低延遲

每 15 分鐘檢查一次 */15 * * * * 普通量

每小時 0 * * * * 標準 (建議採用)

每 6 小時 0 */6 * * * 少量、批次處理

每日 0 0 * * * 歷來資料集合

測試排程器工作

在 Cloud Scheduler 控制台中找出您的工作。

按一下「強制執行」即可手動觸發。

等待幾秒鐘，然後依序前往「Cloud Run」>「Services」(服務) >「cisco-sdwan-log-collector」>「Logs」(記錄)。

確認函式是否已順利執行。

檢查 GCS 值區，確認是否已寫入記錄。

擷取 Google SecOps 服務帳戶

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

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

依序前往「SIEM 設定」>「動態饋給」。

按一下「新增動態消息」。

按一下「設定單一動態饋給」。

在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Cisco SD-WAN logs)。

選取「Google Cloud Storage V2」做為「來源類型」。

選取「Cisco vManage SD-WAN」做為「記錄類型」。

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

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

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

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

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

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

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

按一下 bucket 名稱。

前往「權限」分頁標籤。

按一下「授予存取權」。

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

新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。

指派角色：選取「Storage 物件檢視者」。

按一下 [儲存]。

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

在 Google SecOps 中設定資訊提供，擷取 Cisco vManage SD-WAN 記錄

依序前往「SIEM 設定」>「動態饋給」。

按一下「新增動態消息」。

按一下「設定單一動態饋給」。

在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Cisco SD-WAN logs)。

選取「Google Cloud Storage V2」做為「來源類型」。

選取「Cisco vManage SD-WAN」做為「記錄類型」。

點選 [下一步]。

指定下列輸入參數的值：

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

gs://cisco-sdwan-logs-bucket/cisco-sdwan/

取代：

cisco-sdwan-logs-bucket：您的 GCS bucket 名稱。

cisco-sdwan/：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。

範例：

根層級 bucket：gs://company-logs/

前置字串：gs://company-logs/cisco-sdwan/

有子資料夾：gs://company-logs/cisco-sdwan/audit_logs/

注意： URI 結尾一律要加上尾端斜線 (/)。

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

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

刪除已轉移的檔案：成功轉移檔案後刪除檔案。

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

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

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

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

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

點選 [下一步]。

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

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

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

變數名稱	範例值
`GCS_BUCKET`	`cisco-sdwan-logs-bucket`
`GCS_PREFIX`	`cisco-sdwan/`
`STATE_KEY`	`cisco-sdwan/state.json`
`VMANAGE_HOST`	`https://your-vmanage-server:8443`
`VMANAGE_USERNAME`	`chronicle-api`
`VMANAGE_PASSWORD`	`your-vmanage-password`

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

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