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 バケットを作成および管理する権限
- GCS バケットの IAM ポリシーを管理する権限
- Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
- Cisco vManage SD-WAN 管理コンソールへの特権アクセス
- API アクセス権限を持つ Cisco vManage ユーザー アカウント
Google Cloud Storage バケットを作成する
- Google Cloud Console に移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前（ cisco-sdwan-logs-bucketなど）を入力します。ロケーション タイプ ニーズに基づいて選択します（リージョン、デュアルリージョン、マルチリージョン）。 ロケーション ロケーションを選択します（例: us-central1）。ストレージ クラス Standard（頻繁にアクセスされるログにおすすめ） アクセス制御 均一（推奨） 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Cisco vManage SD-WAN API の認証情報を収集する
API ユーザー アカウントを作成する
- Cisco vManage 管理コンソールにログインします。
- [Administration] > [Settings] > [Users] に移動します。
- [ユーザーを追加] をクリックします。
- 次の構成の詳細を指定します。
- ユーザー名: API アクセス用のユーザー名を入力します（例:
chronicle-api）。 - パスワード: 安全なパスワードを入力します。
- パスワードを確認: パスワードを再入力します。
- ユーザー グループ: 適切な権限を持つユーザー グループを選択します（次のセクションを参照）。
- ユーザー名: API アクセス用のユーザー名を入力します（例:
- [追加] をクリックします。
次の詳細をコピーして安全な場所に保存します。
- ユーザー名: vManage のユーザー名。
- パスワード: vManage のパスワード。
- vManage ベース URL: vManage サーバーのベース URL（例:
https://your-vmanage-server:8443）。
ユーザー権限を構成する
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"
認証が成功すると、2 番目のコマンドは 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 functions のサービス アカウントを作成する
Cloud Run 関数には、GCS バケットに書き込み、Pub/Sub によって呼び出される権限を持つサービス アカウントが必要です。
サービス アカウントの作成
- GCP Console で、[IAM と管理>サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成の詳細を指定します。
- サービス アカウント名: 「
cisco-sdwan-collector-sa」と入力します。 - サービス アカウントの説明: 「
Service account for Cloud Run function to collect Cisco vManage SD-WAN logs」と入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の目的で必要です。
- Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
- Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
- Cloud Functions 起動元: 関数の呼び出しを許可する
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名をクリックします。
- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成の詳細を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス（例:
cisco-sdwan-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。 - ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス（例:
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成の詳細を指定します。
- トピック ID: 「
cisco-sdwan-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Cisco vManage SD-WAN API からログを取得して GCS に書き込みます。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します（インライン エディタを使用して関数を作成します）。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 cisco-sdwan-log-collectorリージョン GCS バケットと一致するリージョンを選択します（例: us-central1）。ランタイム [Python 3.12] 以降を選択します。 [トリガー（省略可）] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、トピック（
cisco-sdwan-trigger）を選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management（IAM）を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント: サービス アカウントを選択します（
cisco-sdwan-collector-sa）。
- サービス アカウント: サービス アカウントを選択します（
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 GCS_BUCKETcisco-sdwan-logs-bucketGCS_PREFIXcisco-sdwan/STATE_KEYcisco-sdwan/state.jsonVMANAGE_HOSThttps://your-vmanage-server:8443VMANAGE_USERNAMEchronicle-apiVMANAGE_PASSWORDyour-vmanage-password[変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
600秒（10 分）を入力します。
- リクエストのタイムアウト:
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
- メモリ: 512 MiB 以上を選択します。
- CPU: [1] を選択します。
- [完了] をクリックします。
- [リソース] セクションで次の操作を行います。
[実行環境] まで下にスクロールします。
- [デフォルト]（推奨）を選択します。
[リビジョン スケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「
0」と入力します。 - インスタンスの最大数:
100と入力します（または、予想される負荷に基づいて調整します）。
- [インスタンスの最小数] に「
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスを作成すると、インライン コードエディタが自動的に開きます。
関数コードを追加する
- [関数のエントリ ポイント] に「main」と入力します。
インライン コードエディタで、次の 2 つのファイルを作成します。
- 最初のファイル: 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- 2 つ目のファイル: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。
Cloud Scheduler ジョブの作成
Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 cisco-sdwan-log-collector-hourlyリージョン Cloud Run functions と同じリージョンを選択する 周波数 0 * * * *（1 時間ごとに正時）タイムゾーン タイムゾーンを選択します（UTC を推奨）。 ターゲット タイプ Pub/Sub トピック トピックを選択する（ cisco-sdwan-trigger）メッセージ本文 {}（空の JSON オブジェクト）[作成] をクリックします。
スケジュールの頻度のオプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
頻度 CRON 式 ユースケース 5 分毎 */5 * * * *大容量、低レイテンシ 15 分ごと */15 * * * *検索量が普通 1 時間ごと 0 * * * *標準（推奨） 6 時間ごと 0 */6 * * *少量、バッチ処理 毎日 0 0 * * *履歴データの収集
スケジューラ ジョブをテストする
- Cloud Scheduler コンソールで、ジョブを見つけます。
- [強制実行] をクリックして手動でトリガーします。
- 数秒待ってから、[Cloud Run> サービス> cisco-sdwan-log-collector > ログ] に移動します。
- 関数が正常に実行されたことを確認します。
- GCS バケットをチェックして、ログが書き込まれたことを確認します。
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
サービス アカウントのメールアドレスを取得する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します（例:
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 サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
- バケット名をクリックします。
- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成の詳細を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [ストレージ オブジェクト閲覧者] を選択します。
[保存] をクリックします。
Cisco vManage SD-WAN のログを取り込むように Google SecOps でフィードを構成する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します（例:
Cisco SD-WAN logs）。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [Cisco vManage SD-WAN] を選択します。
- [次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://cisco-sdwan-logs-bucket/cisco-sdwan/次のように置き換えます。
cisco-sdwan-logs-bucket: GCS バケット名。cisco-sdwan/: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
例:
- ルートバケット:
gs://company-logs/ - 接頭辞あり:
gs://company-logs/cisco-sdwan/ - サブフォルダあり:
gs://company-logs/cisco-sdwan/audit_logs/
- ルートバケット:
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません（テストにおすすめ）。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
アセットの名前空間: アセットの名前空間。
Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
ご不明な点がございましたら、コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。