本頁面由 Cloud Translation API 翻譯而成。

收集 Cisco vManage SD-WAN 記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Amazon S3，將 Cisco vManage SD-WAN 記錄擷取至 Google Security Operations。

事前準備

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

Google SecOps 執行個體
Cisco vManage SD-WAN 管理主控台的特殊存取權
AWS 的具備權限存取權 (S3、IAM、Lambda、EventBridge)

收集 Cisco vManage SD-WAN 必要條件 (憑證和基本網址)

登入 Cisco vManage 管理控制台。
依序前往「管理」>「設定」>「使用者」。
建立新使用者，或使用具備 API 存取權限的現有管理員使用者。
複製下列詳細資料並儲存於安全位置：
- 使用者名稱
- 密碼
- vManage 基準網址 (例如 https://your-vmanage-server:8443)

為 Google SecOps 設定 AWS S3 值區和 IAM

按照這份使用者指南建立 Amazon S3 值區：建立值區
儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 cisco-sdwan-logs-bucket)。
按照這份使用者指南建立使用者：建立 IAM 使用者。
選取建立的「使用者」。
選取「安全憑證」分頁標籤。
在「Access Keys」部分中，按一下「Create Access Key」。
選取「第三方服務」做為「用途」。
點選「下一步」。
選用：新增說明標記。
按一下「建立存取金鑰」。
按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
按一下 [完成]。
選取 [權限] 分頁標籤。
在「Permissions policies」(權限政策) 區段中，按一下「Add permissions」(新增權限)。
選取「新增權限」。
選取「直接附加政策」
搜尋並選取 AmazonS3FullAccess 政策。
點選「下一步」。
按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

在 AWS 控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。

輸入下列政策：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::cisco-sdwan-logs-bucket/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::cisco-sdwan-logs-bucket/cisco-sdwan/state.json"
    }
  ]
}

如果您輸入的值區名稱不同，請替換 cisco-sdwan-logs-bucket。

依序點選「Next」>「Create policy」。
依序前往「IAM」>「角色」。
依序點選「建立角色」>「AWS 服務」>「Lambda」。
附加新建立的政策。
為角色命名 cisco-sdwan-lambda-role，然後按一下「建立角色」。

建立 Lambda 函式

在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
按一下「從頭開始撰寫」。
請提供下列設定詳細資料：

設定值

名稱 cisco-sdwan-log-collector

執行階段 Python 3.13

架構 x86_64

執行角色 cisco-sdwan-lambda-role

設定	值
名稱	`cisco-sdwan-log-collector`
執行階段	Python 3.13
架構	x86_64
執行角色	`cisco-sdwan-lambda-role`

建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (cisco-sdwan-log-collector.py)：

import json
import boto3
import os
import urllib3
import urllib.parse
from datetime import datetime, timezone
from botocore.exceptions import ClientError

# Disable SSL warnings for self-signed certificates
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Environment variables
VMANAGE_HOST = os.environ['VMANAGE_HOST']
VMANAGE_USERNAME = os.environ['VMANAGE_USERNAME']  
VMANAGE_PASSWORD = os.environ['VMANAGE_PASSWORD']
S3_BUCKET = os.environ['S3_BUCKET']
S3_PREFIX = os.environ['S3_PREFIX']
STATE_KEY = os.environ['STATE_KEY']

s3_client = boto3.client('s3')
http = urllib3.PoolManager(cert_reqs='CERT_NONE')

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"
            login_data = urllib.parse.urlencode({
                'j_username': self.username,
                'j_password': self.password
            })

            response = http.request(
                'POST',
                login_url,
                body=login_data,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
                timeout=30
            )

            # Check if login was successful (vManage returns HTML on failure)
            if b'<html>' in response.data or response.status != 200:
                raise Exception("Authentication failed")

            # 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'):
                raise Exception("Failed to get JSESSIONID")

            # 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, timeout=30)

            if response.status == 200:
                self.token = response.data.decode('utf-8')
                return True
            else:
                raise Exception(f"Failed to get XSRF token: {response.status}")

        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()

            # Build query for recent logs
            query = {
                "query": {
                    "condition": "AND",
                    "rules": []
                },
                "size": 10000
            }

            # Add timestamp filter if provided
            if last_timestamp:
                # Convert timestamp to epoch milliseconds for vManage API
                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:
                # Get last 1 hour of logs by default
                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,
                timeout=60
            )

            if response.status == 200:
                return json.loads(response.data.decode('utf-8'))
            else:
                print(f"Failed to get audit logs: {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()

            # Build query for recent alarms
            query = {
                "query": {
                    "condition": "AND",
                    "rules": []
                },
                "size": 10000
            }

            # Add timestamp filter if provided
            if last_timestamp:
                # Convert timestamp to epoch milliseconds for vManage API
                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:
                # Get last 1 hour of alarms by default
                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,
                timeout=60
            )

            if response.status == 200:
                return json.loads(response.data.decode('utf-8'))
            else:
                print(f"Failed to get alarms: {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()

            # Build query for recent events
            query = {
                "query": {
                    "condition": "AND",
                    "rules": []
                },
                "size": 10000
            }

            # Add timestamp filter if provided  
            if last_timestamp:
                # Convert timestamp to epoch milliseconds for vManage API
                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:
                # Get last 1 hour of events by default
                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,
                timeout=60
            )

            if response.status == 200:
                return json.loads(response.data.decode('utf-8'))
            else:
                print(f"Failed to get events: {response.status}")
                return None

        except Exception as e:
            print(f"Error getting events: {e}")
            return None

def get_last_run_time():
    """Get the last successful run timestamp from S3"""
    try:
        response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        state_data = json.loads(response['Body'].read())
        return state_data.get('last_run_time')
    except ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchKey':
            print("No previous state found, collecting last hour of logs")
            return None
        else:
            print(f"Error reading state: {e}")
            return None
    except Exception as e:
        print(f"Error reading state: {e}")
        return None

def update_last_run_time(timestamp):
    """Update the last successful run timestamp in S3"""
    try:
        state_data = {
            'last_run_time': timestamp,
            'updated_at': datetime.now(timezone.utc).isoformat()
        }

        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=STATE_KEY,
            Body=json.dumps(state_data),
            ContentType='application/json'
        )

        print(f"Updated state with timestamp: {timestamp}")

    except Exception as e:
        print(f"Error updating state: {e}")

def upload_logs_to_s3(logs_data, log_type, timestamp):
    """Upload logs to S3 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

        # Create filename with timestamp
        dt = datetime.now(timezone.utc)
        filename = f"{S3_PREFIX}{log_type}/{dt.strftime('%Y/%m/%d')}/{log_type}_{dt.strftime('%Y%m%d_%H%M%S')}.json"

        # Upload to S3
        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=filename,
            Body=json.dumps(logs_data),
            ContentType='application/json'
        )

        print(f"Uploaded {len(logs_data['data'])} {log_type} records to s3://{S3_BUCKET}/{filename}")

    except Exception as e:
        print(f"Error uploading {log_type} to S3: {e}")

def lambda_handler(event, context):
    """Main Lambda handler function"""
    print(f"Starting Cisco vManage log collection at {datetime.now(timezone.utc)}")

    try:
        # Get last run time
        last_run_time = get_last_run_time()

        # Initialize vManage API client
        vmanage = VManageAPI(VMANAGE_HOST, VMANAGE_USERNAME, VMANAGE_PASSWORD)

        # Authenticate
        if not vmanage.authenticate():
            return {
                'statusCode': 500,
                'body': json.dumps('Failed to authenticate with vManage')
            }

        print("Successfully authenticated with vManage")

        # 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_s3(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(current_time)

        print(f"Collection completed. Total records processed: {total_records}")

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Log collection completed successfully',
                'total_records': total_records,
                'timestamp': datetime.now(timezone.utc).isoformat()
            })
        }

    except Exception as e:
        print(f"Lambda execution error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps(f'Error: {str(e)}')
        }

依序前往「設定」>「環境變數」。
依序點選「編輯」> 新增環境變數。

輸入下列環境變數，並將換成您的值：

鍵	範例值
`S3_BUCKET`	`cisco-sdwan-logs-bucket`
`S3_PREFIX`	`cisco-sdwan/`
`STATE_KEY`	`cisco-sdwan/state.json`
`VMANAGE_HOST`	`https://your-vmanage-server:8443`
`VMANAGE_USERNAME`	`your-vmanage-username`
`VMANAGE_PASSWORD`	`your-vmanage-password`

建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「cisco-sdwan-log-collector」)。
選取「設定」分頁標籤。
在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

注意： 逾時是上限，也就是說，如果函式在計時器達到 5 分鐘時仍在執行，AWS Lambda 就會終止函式。如果執行時間較短，系統會正常完成作業，並只收取實際執行時間的費用。

建立 EventBridge 排程

依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
提供下列設定詳細資料：
- 週期性時間表：費率 (1 hour)
- 目標：您的 Lambda 函式 cisco-sdwan-log-collector
- Name (名稱)：cisco-sdwan-log-collector-1h
按一下「建立時間表」。

選用：為 Google SecOps 建立唯讀 IAM 使用者和金鑰

在 AWS 控制台中，依序前往「IAM」>「Users」>「Add users」。
點選 [Add users] (新增使用者)。
提供下列設定詳細資料：
- 使用者：secops-reader
- 存取類型：存取金鑰 - 程式輔助存取
按一下「建立使用者」。
附加最低讀取權限政策 (自訂)：依序選取「使用者」>「secops-reader」>「權限」>「新增權限」>「直接附加政策」>「建立政策」。

在 JSON 編輯器中輸入下列政策：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::cisco-sdwan-logs-bucket",
      "Condition": {
        "StringLike": {
          "s3:prefix": ["cisco-sdwan/*"]
        }
      }
    },
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::cisco-sdwan-logs-bucket/cisco-sdwan/*"
    }
  ]
}

將政策命名為 secops-reader-policy。
按一下「建立政策」。
返回使用者建立頁面，搜尋並選取 secops-reader-policy。
按一下「下一步：代碼」。
按一下 [下一步：檢閱]。
按一下「建立使用者」。
下載 CSV (這些值會輸入至動態饋給)。

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

依序前往「SIEM 設定」>「動態饋給」。
按一下「+ 新增動態消息」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Cisco SD-WAN logs)。
選取「Amazon S3 V2」做為「來源類型」。
選取「Cisco vManage SD-WAN」做為「記錄類型」。
點選「下一步」。
指定下列輸入參數的值：
- S3 URI：s3://cisco-sdwan-logs-bucket/cisco-sdwan/
- 來源刪除選項：根據偏好選取刪除選項。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：套用至這個動態饋給事件的標籤。
點選「下一步」。
在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

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