Recolha registos de monitorização contínua do Qualys
Este código do analisador Logstash extrai primeiro campos como o IP de origem, o utilizador, o método e o protocolo de aplicação de mensagens de registo não processadas através de padrões grok. Em seguida, mapeia campos específicos dos dados de registo não processados para os respetivos campos no modelo de dados unificado (UDM), faz conversões de tipos de dados e enriquece os dados com etiquetas e metadados adicionais antes de estruturar finalmente o resultado no formato UDM pretendido.
Antes de começar
Certifique-se de que tem os seguintes pré-requisitos:
- Instância do Google Security Operations.
- Acesso privilegiado a Google Cloud.
- Acesso privilegiado ao Qualys.
Ative as APIs necessárias:
- Inicie sessão na Google Cloud consola.
- Aceda a APIs e serviços > Biblioteca.
- Pesquise as seguintes APIs e ative-as:
- Cloud Functions API
- API do Cloud Scheduler
- Cloud Pub/Sub (necessário para o Cloud Scheduler invocar funções)
Crie um Google Cloud contentor de armazenamento
- Inicie sessão na Google Cloud consola.
Aceda à página Contentores do Cloud Storage.
Clique em Criar.
Configure o contentor:
- Nome: introduza um nome exclusivo que cumpra os requisitos do nome do contentor (por exemplo, qualys-asset-bucket).
- Escolha onde quer armazenar os seus dados: selecione uma localização.
- Escolha uma classe de armazenamento para os seus dados: selecione uma classe de armazenamento predefinida para o contentor ou selecione Autoclass para a gestão automática da classe de armazenamento.
- Escolha como controlar o acesso aos objetos: selecione não para aplicar a prevenção de acesso público e selecione um modelo de controlo de acesso para os objetos do seu contentor.
- Classe de armazenamento: escolha com base nas suas necessidades (por exemplo, Standard).
Clique em Criar.
Crie uma conta de serviço do Google Cloud
- Inicie sessão na Google Cloud consola.
- Aceda a IAM e administrador > Contas de serviço.
- Crie uma nova conta de serviço.
- Atribua-lhe um nome descritivo (por exemplo, qualys-user).
- Conceda à conta de serviço a função de administrador de objetos de armazenamento no contentor do GCS que criou no passo anterior.
- Conceda à conta de serviço a função de invocador do Cloud Functions.
- Crie uma chave SSH para a conta de serviço.
- Transfira um ficheiro de chave JSON para a conta de serviço. Mantenha este ficheiro seguro.
Opcional: crie um utilizador da API dedicado no Qualys
- Inicie sessão na consola do Qualys.
- Aceda a Utilizadores.
- Clique em Novo > Utilizador.
- Introduza as informações gerais necessárias para o utilizador.
- Selecione o separador Função do utilizador.
- Certifique-se de que a caixa de verificação Acesso à API está selecionada para a função.
- Clique em Guardar.
Identifique o URL da API Qualys específico
Opção 1
Identifique os seus URLs conforme mencionado na identificação da plataforma.
Opção 2
- Inicie sessão na consola do Qualys.
- Aceda a Ajuda > Acerca de.
- Desloque a página para ver estas informações em Centro de operações de segurança (SOC).
- Copie o URL da API Qualys.
Configure a função do Google Cloud
- Aceda ao Cloud Functions na Google Cloud consola.
- Clique em Criar função.
Configure a função:
- Nome: introduza um nome para a sua função (por exemplo, fetch-qualys-cm-alerts).
- Região: selecione uma região próxima do seu contentor.
- Tempo de execução: Python 3.10 (ou o tempo de execução preferido).
- Acionador: escolha o acionador HTTP, se necessário, ou o Cloud Pub/Sub para execução agendada.
- Autenticação: proteja com autenticação.
- Escreva o código com um editor inline:
```python from google.cloud import storage import requests import base64 import json # Google Cloud Storage Configuration BUCKET_NAME = "<bucket-name>" FILE_NAME = "qualys_cm_alerts.json" # Qualys API Credentials QUALYS_USERNAME = "<qualys-username>" QUALYS_PASSWORD = "<qualys-password>" QUALYS_BASE_URL = "https://<qualys_base_url>" def fetch_cm_alerts(): """Fetch alerts from Qualys Continuous Monitoring.""" auth = base64.b64encode(f"{QUALYS_USERNAME}:{QUALYS_PASSWORD}".encode()).decode() headers = { "Authorization": f"Basic {auth}", "Content-Type": "application/xml" } payload = """ <ServiceRequest> <filters> <Criteria field="alert.date" operator="GREATER">2024-01-01</Criteria> </filters> </ServiceRequest> """ response = requests.post(f"{QUALYS_BASE_URL}/qps/rest/2.0/search/cm/alert", headers=headers, data=payload) response.raise_for_status() return response.json() def upload_to_gcs(data): """Upload data to Google Cloud Storage.""" client = storage.Client() bucket = client.get_bucket(BUCKET_NAME) blob = bucket.blob(FILE_NAME) blob.upload_from_string(json.dumps(data, indent=2), content_type="application/json") def main(request): """Cloud Function entry point.""" try: alerts = fetch_cm_alerts() upload_to_gcs(alerts) return "Qualys CM alerts uploaded to Cloud Storage successfully!" except Exception as e: return f"An error occurred: {e}", 500 ```
Clique em Implementar após concluir a configuração.
Configure o Cloud Scheduler
- Aceda ao Cloud Scheduler na Google Cloud consola.
- Clique em Criar tarefa.
Configure a tarefa:
- Nome: introduza um nome para a tarefa (por exemplo, trigger-fetch-qualys-cm-alerts).
- Frequência: use a sintaxe cron para especificar a programação (por exemplo,
0 * * * *
para executar a cada hora). - Fuso horário: defina o seu fuso horário preferido.
- Tipo de acionador: escolha HTTP.
- URL do acionador: introduza o URL da função na nuvem (encontrado nos detalhes da função após a implementação).
- Método: escolha POST.
Crie o trabalho.
Configure feeds
Para configurar um feed, siga estes passos:
- Aceda a Definições do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Na página seguinte, clique em Configurar um único feed.
- No campo Nome do feed, introduza um nome para o feed; por exemplo, Registos de monitorização contínua do Qualys.
- Selecione Google Cloud Storage V2 como Tipo de origem.
- Selecione Qualys Continuous Monitoring como o Tipo de registo.
- Clicar em Seguinte.
Especifique valores para os seguintes parâmetros de entrada:
- URI do contentor de armazenamento: o Google Cloud URI de origem do contentor de armazenamento.
- Opções de eliminação da origem: selecione a opção de eliminação de acordo com a sua preferência.
Clicar em Seguinte.
Reveja a nova configuração do feed no ecrã Finalizar e, de seguida, clique em Enviar.
Tabela de mapeamento da UDM
Campo de registo | Mapeamento de UDM | Lógica |
---|---|---|
Alert.alertInfo.appVersion | metadata.product_version | Mapeado diretamente de Alert.alertInfo.appVersion |
Alert.alertInfo.operatingSystem | principal.platform_version | Mapeado diretamente de Alert.alertInfo.operatingSystem |
Alert.alertInfo.port | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.port e adicionado como um par de chave-valor em additional.fields com a chave "Porta de alerta" |
Alert.alertInfo.protocol | network.ip_protocol | Mapeado diretamente de Alert.alertInfo.protocol |
Alert.alertInfo.sslIssuer | network.tls.client.certificate.issuer | Mapeado diretamente de Alert.alertInfo.sslIssuer |
Alert.alertInfo.sslName | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.sslName e adicionado como um par de chave-valor em additional.fields com a chave "Nome SSL" |
Alert.alertInfo.sslOrg | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.sslOrg e adicionado como um par de chave-valor em additional.fields com a chave "SSL Org" |
Alert.alertInfo.ticketId | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.ticketId e adicionado como um par de chave-valor em additional.fields com a chave "ID do pedido" |
Alert.alertInfo.vpeConfidence | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.vpeConfidence e adicionado como um par de chave-valor em additional.fields com a chave "VPE Confidence" |
Alert.alertInfo.vpeStatus | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.alertInfo.vpeStatus e adicionado como um par de chave-valor em additional.fields com a chave "VPE Confidence" |
Alert.eventType | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.eventType e adicionado como um par de chave-valor em additional.fields com a chave "Tipo de evento" |
Alert.hostname | principal.hostname | Mapeado diretamente de Alert.hostname |
Alert.id | security_result.threat_id | Mapeado diretamente de Alert.id |
Alert.ipAddress | principal.ip | Mapeado diretamente de Alert.ipAddress |
Alert.profile.id | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.profile.id e adicionado como um par de chave-valor em additional.fields com a chave "ID do perfil" |
Alert.profile.title | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.profile.title e adicionado como um par de chave-valor em additional.fields com a chave "Título do perfil" |
Alert.qid | vulnerability.name | Mapeado como "QID: Alert.qid |
Alert.source | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.source e adicionado como um par de chave-valor em additional.fields com a chave "Origem do alerta" |
Alert.triggerUuid | metadata.product_log_id | Mapeado diretamente de Alert.triggerUuid |
Alert.vulnCategory | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.vulnCategory e adicionado como um par de chave-valor em additional.fields com a chave "Categoria de vulnerabilidade" |
Alert.vulnSeverity | vulnerability.severity | Mapeada com base no valor de Alert.vulnSeverity : 1 a 3: BAIXO, 4 a 6: MÉDIO, 7 a 8: ELEVADO |
Alert.vulnTitle | vulnerability.description | Mapeado diretamente de Alert.vulnTitle |
Alert.vulnType | additional.fields.value.string_value | Mapeado diretamente a partir de Alert.vulnType e adicionado como um par de chave-valor em additional.fields com a chave "Vulnerability Type" |
Anfitrião | principal.ip | Analisado a partir da linha de registo "Host: |
edr.client.ip_addresses | Copiados do principal.ip |
|
edr.client.hostname | Copiados do principal.hostname |
|
edr.raw_event_name | Definido como "STATUS_UPDATE" se Alert.ipAddress , Alert.hostname ou src_ip estiverem presentes. Caso contrário, definido como "GENERIC_EVENT" |
|
metadata.event_timestamp | Extraído dos campos Alert.eventDate ou timestamp . A Alert.eventDate tem prioridade se existir. Caso contrário, é usada a timestamp . A data/hora é convertida em UTC. |
|
metadata.event_type | A mesma lógica que edr.raw_event_name |
|
metadata.log_type | Definido como "QUALYS_CONTINUOUS_MONITORING" | |
metadata.product_name | Definido como "QUALYS_CONTINUOUS_MONITORING" | |
metadata.vendor_name | Definido como "QUALYS_CONTINUOUS_MONITORING" | |
network.application_protocol | Analisado a partir da linha de registo " |
|
network.http.method | Analisado a partir da linha de registo " |
|
timestamp | event.timestamp | Extraído dos campos Alert.eventDate ou timestamp . A Alert.eventDate tem prioridade se existir. Caso contrário, é usada a timestamp . A data/hora é convertida em UTC. |
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais da Google SecOps.