Recopilar registros de Cisco vManage SD-WAN

Disponible en:

En este documento se explica cómo ingerir registros de SD-WAN de Cisco vManage en Google Security Operations mediante Amazon S3.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Acceso con privilegios a la consola de gestión Cisco vManage SD-WAN
  • Acceso privilegiado a AWS (S3, IAM, Lambda y EventBridge)

Recoger los requisitos previos de Cisco vManage SD-WAN (credenciales y URL base)

  1. Inicia sesión en la consola de gestión de Cisco vManage.
  2. Ve a Administración > Configuración > Usuarios.
  3. Crea un usuario o usa un usuario administrador que ya tenga privilegios de acceso a la API.
  4. Copia y guarda en un lugar seguro los siguientes detalles:
    • Username (Nombre de usuario)
    • Password (Contraseña)
    • URL base de vManage (por ejemplo, https://your-vmanage-server:8443)

Configurar un segmento de AWS S3 y IAM para Google SecOps

  1. Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
  2. Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, cisco-sdwan-logs-bucket).
  3. Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
  4. Selecciona el Usuario creado.
  5. Selecciona la pestaña Credenciales de seguridad.
  6. En la sección Claves de acceso, haz clic en Crear clave de acceso.
  7. Selecciona Servicio de terceros como Caso práctico.
  8. Haz clic en Siguiente.
  9. Opcional: añade una etiqueta de descripción.
  10. Haz clic en Crear clave de acceso.
  11. Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
  12. Haz clic en Listo.
  13. Selecciona la pestaña Permisos.
  14. En la sección Políticas de permisos, haz clic en Añadir permisos.
  15. Selecciona Añadir permisos.
  16. Seleccione Adjuntar políticas directamente.
  17. Busca y selecciona la política AmazonS3FullAccess.
  18. Haz clic en Siguiente.
  19. Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas de S3

  1. En la consola de AWS, ve a IAM > Policies > Create policy > pestaña JSON.

  2. Introduce la siguiente política:

    {
  "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"
    }
  ]
}
    • Sustituye cisco-sdwan-logs-bucket si has introducido otro nombre de segmento.

  3. Haz clic en Siguiente > Crear política.

  4. Ve a IAM > Roles.

  5. Haga clic en Crear rol > Servicio de AWS > Lambda.

  6. Adjunte la política que acaba de crear.

  7. Dale el nombre cisco-sdwan-lambda-role al rol y haz clic en Crear rol.

Crear la función Lambda

  1. En la consola de AWS, ve a Lambda > Funciones > Crear función.
  2. Haz clic en Crear desde cero.

  3. Proporciona los siguientes detalles de configuración:

    Ajuste Valor
    Nombre cisco-sdwan-log-collector
    Tiempo de ejecución Python 3.13
    Arquitectura x86_64
    Rol de ejecución cisco-sdwan-lambda-role

  4. Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (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)}')
        }

  5. Vaya a Configuración > Variables de entorno.

  6. Haz clic en Editar > Añadir nueva variable de entorno.

  7. Introduce las siguientes variables de entorno y sustituye los valores por los tuyos:

    Clave Valor de ejemplo
    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

  8. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > cisco-sdwan-log-collector).

  9. Seleccione la pestaña Configuración.

  10. En el panel Configuración general, haz clic en Editar.

  11. Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Crear una programación de EventBridge

  1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
  2. Proporcione los siguientes detalles de configuración:
    • Programación periódica: tarifa (1 hour)
    • Destino: tu función Lambda cisco-sdwan-log-collector
    • Nombre: cisco-sdwan-log-collector-1h
  3. Haz clic en Crear programación.

Opcional: Crear un usuario y claves de gestión de identidades y accesos de solo lectura para Google SecOps

  1. En la consola de AWS, ve a IAM > Usuarios > Añadir usuarios.
  2. Haz clic en Add users (Añadir usuarios).
  3. Proporcione los siguientes detalles de configuración:
    • Usuario: secops-reader
    • Tipo de acceso: clave de acceso (acceso programático)
  4. Haz clic en Crear usuario.
  5. Asigna una política de lectura mínima (personalizada): Usuarios > secops-reader > Permisos > Añadir permisos > Asignar políticas directamente > Crear política.

  6. En el editor de JSON, introduce la siguiente política:

    {
  "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/*"
    }
  ]
}

  7. Asigna un nombre a la política secops-reader-policy.

  8. Haz clic en Crear política.

  9. Vuelve a la creación de usuarios, busca y selecciona secops-reader-policy.

  10. Haz clic en Siguiente: Etiquetas.

  11. Haz clic en Siguiente: Revisar.

  12. Haz clic en Crear usuario.

  13. Descarga el archivo CSV (estos valores se introducen en el feed).

Configurar un feed en Google SecOps para ingerir registros de Cisco vManage SD-WAN

  1. Ve a Configuración de SIEM > Feeds.
  2. Haz clic en + Añadir nuevo feed.
  3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, Cisco SD-WAN logs).
  4. Selecciona Amazon S3 V2 como Tipo de fuente.
  5. Selecciona Cisco vManage SD-WAN como Tipo de registro.
  6. Haz clic en Siguiente.
  7. Especifique los valores de los siguientes parámetros de entrada:
    • URI de S3: s3://cisco-sdwan-logs-bucket/cisco-sdwan/
    • Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
    • Antigüedad máxima del archivo: incluye los archivos modificados en los últimos días. Valor predeterminado: 180 días.
    • ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
    • Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
    • Espacio de nombres de recursos: el espacio de nombres de recursos.
    • Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
  8. Haz clic en Siguiente.
  9. Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.