Coletar registros da SD-WAN do Cisco vManage
Este documento explica como ingerir registros de SD-WAN do Cisco vManage no Google Security Operations usando o Google Cloud Storage. O Cisco vManage SD-WAN é um sistema centralizado de gerenciamento de rede que oferece visibilidade e controle sobre a estrutura SD-WAN. Assim, os administradores podem monitorar o desempenho da rede, configurar políticas e gerenciar a segurança em redes corporativas distribuídas.
Antes de começar
Verifique se você atende os seguintes pré-requisitos:
- Uma instância do Google SecOps
- Um projeto do GCP com a API Cloud Storage ativada
- Permissões para criar e gerenciar buckets do GCS
- Permissões para gerenciar políticas do IAM em buckets do GCS
- Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
- Acesso privilegiado ao console de gerenciamento do Cisco vManage SD-WAN
- Conta de usuário do Cisco vManage com permissões de acesso à API
Criar um bucket do Google Cloud Storage
- Acesse o Console do Google Cloud.
- Selecione seu projeto ou crie um novo.
- No menu de navegação, acesse Cloud Storage > Buckets.
- Clique em Criar bucket.
Informe os seguintes detalhes de configuração:
Configuração Valor Nomeie seu bucket Insira um nome exclusivo globalmente, por exemplo, cisco-sdwan-logs-bucket.Tipo de local Escolha com base nas suas necessidades (região, birregional, multirregional) Local Selecione o local (por exemplo, us-central1).Classe de armazenamento Padrão (recomendado para registros acessados com frequência) Controle de acesso Uniforme (recomendado) Ferramentas de proteção Opcional: ativar o controle de versões de objetos ou a política de retenção Clique em Criar.
Coletar credenciais da API SD-WAN do Cisco vManage
Criar conta de usuário da API
- Faça login no Cisco vManage Management Console.
- Acesse Administração > Configurações > Usuários.
- Clique em Adicionar usuário.
- Informe os seguintes detalhes de configuração:
- Nome de usuário: insira um nome de usuário para acesso à API (por exemplo,
chronicle-api). - Senha: insira uma senha forte.
- Confirmar senha: digite a senha novamente.
- Grupo de usuários: selecione um grupo de usuários com as permissões adequadas (consulte a próxima seção).
- Nome de usuário: insira um nome de usuário para acesso à API (por exemplo,
- Clique em Adicionar.
Copie e salve em um local seguro os seguintes detalhes:
- Nome de usuário: seu nome de usuário do vManage.
- Senha: sua senha do vManage.
- URL de base do vManage: o URL de base do seu servidor vManage (por exemplo,
https://your-vmanage-server:8443).
Configurar permissões de usuário
A conta de usuário da API precisa de permissões específicas para acessar registros de auditoria, alarmes e eventos.
- No Cisco vManage Management Console, acesse Administração > Configurações > Grupos de usuários.
- Selecione o grupo de usuários atribuído ao usuário da API ou crie um novo grupo.
- Clique em Editar.
- Na seção Recurso, verifique se as seguintes permissões estão ativadas:
- Registro de auditoria: selecione a permissão Ler.
- Alarmes: selecione a permissão Ler.
- Eventos: selecione a permissão Leitura.
- Clique em Atualizar.
Verificar o acesso à API
Teste suas credenciais antes de prosseguir com a integração:
- Abra um terminal ou prompt de comando.
Execute o seguinte comando para testar a autenticação:
# 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"
Se a autenticação for bem-sucedida, o segundo comando vai retornar uma string de 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.
Criar uma conta de serviço para a função do Cloud Run
A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS e ser invocada pelo Pub/Sub.
Criar conta de serviço
- No Console do GCP, acesse IAM e administrador > Contas de serviço.
- Clique em Criar conta de serviço.
- Informe os seguintes detalhes de configuração:
- Nome da conta de serviço: insira
cisco-sdwan-collector-sa. - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect Cisco vManage SD-WAN logs.
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
- Clique em Selecionar papel.
- Pesquise e selecione Administrador de objetos do Storage.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Run.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Functions.
- Clique em Continuar.
- Clique em Concluído.
Esses papéis são necessários para:
- Administrador de objetos do Storage: grava registros em um bucket do GCS e gerencia arquivos de estado.
- Invocador do Cloud Run: permite que o Pub/Sub invoque a função
- Invocador do Cloud Functions: permite a invocação de funções
Conceder permissões do IAM no bucket do GCS
Conceda permissões de gravação à conta de serviço no bucket do GCS:
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket.
- Acesse a guia Permissões.
- Clique em Conceder acesso.
- Informe os seguintes detalhes de configuração:
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
cisco-sdwan-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Atribuir papéis: selecione Administrador de objetos do Storage.
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
- Clique em Salvar.
Criar tópico Pub/Sub
Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.
- No Console do GCP, acesse Pub/Sub > Tópicos.
- Selecione Criar tópico.
- Informe os seguintes detalhes de configuração:
- ID do tópico: insira
cisco-sdwan-trigger. - Não altere as outras configurações.
- ID do tópico: insira
- Clique em Criar.
Criar uma função do Cloud Run para coletar registros
A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API SD-WAN do Cisco vManage e gravá-los no GCS.
- No console do GCP, acesse o Cloud Run.
- Clique em Criar serviço.
- Selecione Função (use um editor in-line para criar uma função).
Na seção Configurar, forneça os seguintes detalhes de configuração:
Configuração Valor Nome do serviço cisco-sdwan-log-collectorRegião Selecione a região que corresponde ao seu bucket do GCS (por exemplo, us-central1).Ambiente de execução Selecione Python 3.12 ou uma versão mais recente. Na seção Acionador (opcional):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico (
cisco-sdwan-trigger). - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Confira o Identity and Access Management (IAM).
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione a conta de serviço (
cisco-sdwan-collector-sa).
- Conta de serviço: selecione a conta de serviço (
Acesse a guia Contêineres:
- Clique em Variáveis e secrets.
- Clique em + Adicionar variável para cada variável de ambiente:
Nome da variável Valor de exemplo GCS_BUCKETcisco-sdwan-logs-bucketGCS_PREFIXcisco-sdwan/STATE_KEYcisco-sdwan/state.jsonVMANAGE_HOSThttps://your-vmanage-server:8443VMANAGE_USERNAMEchronicle-apiVMANAGE_PASSWORDyour-vmanage-passwordRole a tela para baixo na guia Variáveis e secrets até Solicitações:
- Tempo limite da solicitação: insira
600segundos (10 minutos).
- Tempo limite da solicitação: insira
Acesse a guia Configurações em Contêineres:
- Na seção Recursos:
- Memória: selecione 512 MiB ou mais.
- CPU: selecione 1.
- Clique em Concluído.
- Na seção Recursos:
Role a tela para baixo até Ambiente de execução:
- Selecione Padrão (recomendado).
Na seção Escalonamento de revisão:
- Número mínimo de instâncias: insira
0. - Número máximo de instâncias: insira
100ou ajuste com base na carga esperada.
- Número mínimo de instâncias: insira
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço é criado, o editor de código inline é aberto automaticamente.
Adicionar código da função
- Insira main em Ponto de entrada da função.
No editor de código em linha, crie dois arquivos:
- Primeiro arquivo: 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- Segundo arquivo: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).
Criar o job do Cloud Scheduler
O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.
- No Console do GCP, acesse o Cloud Scheduler.
- Clique em Criar job.
Informe os seguintes detalhes de configuração:
Configuração Valor Nome cisco-sdwan-log-collector-hourlyRegião Selecione a mesma região da função do Cloud Run Frequência 0 * * * *(a cada hora, na hora)Fuso horário Selecione o fuso horário (UTC recomendado) Tipo de destino Pub/Sub Tópico Selecione o assunto ( cisco-sdwan-trigger)Corpo da mensagem {}(objeto JSON vazio)Clique em Criar.
Opções de frequência de programação
Escolha a frequência com base no volume de registros e nos requisitos de latência:
Frequência Expressão Cron Caso de uso A cada 5 minutos */5 * * * *Alto volume e baixa latência A cada 15 minutos */15 * * * *Volume médio A cada hora 0 * * * *Padrão (recomendado) A cada 6 horas 0 */6 * * *Baixo volume, processamento em lote Diário 0 0 * * *Coleta de dados históricos
Testar o job do programador
- No console do Cloud Scheduler, encontre seu job.
- Clique em Forçar execução para acionar manualmente.
- Aguarde alguns segundos e acesse Cloud Run > Serviços > cisco-sdwan-log-collector > Registros.
- Verifique se a função foi executada com sucesso.
- Verifique o bucket do GCS para confirmar se os registros foram gravados.
Recuperar a conta de serviço do Google SecOps
O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.
Receber o e-mail da conta de serviço
- Acesse Configurações do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Cisco SD-WAN logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Cisco vManage SD-WAN como o Tipo de registro.
Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopie esse endereço de e-mail para usar na próxima etapa.
Conceder permissões do IAM à conta de serviço do Google SecOps
A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket.
- Acesse a guia Permissões.
- Clique em Conceder acesso.
- Informe os seguintes detalhes de configuração:
- Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
- Atribuir papéis: selecione Leitor de objetos do Storage.
Clique em Salvar.
Configurar um feed no Google SecOps para ingerir registros de SD-WAN do Cisco vManage
- Acesse Configurações do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Cisco SD-WAN logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Cisco vManage SD-WAN como o Tipo de registro.
- Clique em Próxima.
Especifique valores para os seguintes parâmetros de entrada:
URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:
gs://cisco-sdwan-logs-bucket/cisco-sdwan/Substitua:
cisco-sdwan-logs-bucket: o nome do bucket do GCS.cisco-sdwan/: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
Exemplos:
- Bucket raiz:
gs://company-logs/ - Com prefixo:
gs://company-logs/cisco-sdwan/ - Com subpasta:
gs://company-logs/cisco-sdwan/audit_logs/
- Bucket raiz:
Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:
- Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
- Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.
Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.
Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
Namespace do recurso: o namespace do recurso.
Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.
Clique em Próxima.
Revise a nova configuração do feed na tela Finalizar e clique em Enviar.
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.