Coletar registros de contexto de recursos da Qualys
Este documento explica como ingerir registros de contexto de recursos da Qualys no Google Security Operations usando o Google Cloud Storage.
O contexto de recursos da Qualys fornece dados de inventário de recursos de host (endereços IP, nomes de DNS, sistema operacional, metadados do provedor de nuvem e tags) da plataforma Qualys. Como esses dados são extraídos da API Qualys, uma função do Cloud Run consulta periodicamente a API Qualys Host List, converte a resposta XML em JSON e a grava em um bucket do Cloud Storage, onde o Google SecOps a ingere.
Antes de começar
Verifique se você tem os pré-requisitos a seguir:
- Uma instância do Google SecOps
- Um projeto na nuvem do Google Cloud com as APIs Cloud Storage, Cloud Run, Pub/Sub e Cloud Scheduler ativadas
- Permissões para criar e gerenciar buckets do Cloud Storage, contas de serviço, funções do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
- Uma conta da Qualys com acesso à API (um usuário com a função Gerente ou uma função que tenha acesso à API e visibilidade dos recursos que você quer coletar)
Criar um bucket do Cloud Storage
- Acesse o Console do Google Cloud.
- Selecione ou crie um projeto.
- Acesse Cloud Storage > Buckets.
- Clique em Criar bucket.
- Informe os seguintes detalhes de configuração:
- Nomeie seu bucket: insira um nome globalmente exclusivo, por exemplo,
qualys-asset-context-logs. - Tipo de local: escolha com base nas suas necessidades (região, birregional ou multirregional).
- Local: selecione o local (por exemplo,
us-central1). - Classe de armazenamento: selecione Padrão.
- Controle de acesso: selecione Uniforme.
- Nomeie seu bucket: insira um nome globalmente exclusivo, por exemplo,
- Clique em Criar.
Coletar credenciais da API Qualys
- Faça login no aplicativo da Web Qualys.
- Identifique ou crie um usuário da API:
- Acesse Usuários > Gerenciamento de usuários.
- Use um usuário existente ou clique em Novo > Usuário e atribua a função Administrador (ou uma função com acesso à API).
- Registre o nome de usuário e a senha do usuário.
Identifique o URL do servidor da API do Qualys, que é específico para sua plataforma Qualys (por exemplo,
https://qualysapi.qg3.apps.qualys.com). Para encontrá-lo, consulte Identificar sua plataforma Qualys.
Criar uma conta de serviço para a função do Cloud Run
- 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
qualys-asset-context-sa. - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect Qualys asset context logs.
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Conceda os seguintes papéis:
- Administrador de objetos do Storage: grava registros de host no bucket.
- Invocador do Cloud Run: permite que o Pub/Sub invoque a função.
- Clique em Continuar e depois em Concluído.
Conceder à conta de serviço acesso ao bucket do Cloud Storage
- 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 (
qualys-asset-context-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 (
- Clique em Salvar.
Criar um tópico do Pub/Sub
- Acesse Pub/Sub > Tópicos.
- Selecione Criar tópico.
- No campo ID do tópico, insira
qualys-asset-context-trigger. - Clique em Criar.
Criar a função do Cloud Run para coletar registros
A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler. Ele chama a API de lista de hosts do Qualys FO, converte a resposta XML em JSON e grava os registros de host no bucket.
- Acesse o Cloud Run.
- Clique em Criar serviço.
- Selecione Função para usar o editor de código in-line.
- Informe os seguintes detalhes de configuração:
- Nome do serviço: insira
qualys-asset-context-collector. - Região: selecione a região que corresponde ao seu bucket (por exemplo,
us-central1). - Ambiente de execução: selecione Python 3.12 ou uma versão mais recente.
- Nome do serviço: insira
- Na seção Acionador:
- Clique em Adicionar acionador.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, selecione
qualys-asset-context-trigger. - Clique em Salvar.
- Na seção Autenticação, selecione Exigir autenticação.
- Abra Contêineres, rede, segurança e, na guia Segurança, defina Conta de serviço como
qualys-asset-context-sa. Na guia Contêineres > Variáveis e secrets, adicione as seguintes variáveis de ambiente:
Nome da variável Valor de exemplo GCS_BUCKETqualys-asset-context-logsGCS_PREFIXqualys-asset-context/QUALYS_BASE_URLhttps://qualysapi.qg3.apps.qualys.comQUALYS_USERNAMEyour-qualys-usernameQUALYS_PASSWORDyour-qualys-passwordTRUNCATION_LIMIT1000HTTP_TIMEOUT300MAX_PAGES50Defina o Tempo limite da solicitação como
600segundos e a Memória como 512 MiB ou mais.Clique em Criar. Depois que o serviço é criado, o editor de código inline é aberto.
Adicionar o código da função
- No campo Ponto de entrada da função, insira
main. No editor de código inline, substitua o conteúdo de
main.pypelo seguinte código:import functions_framework from google.cloud import storage import os import time import uuid import json import base64 import requests import xmltodict storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'qualys-asset-context/') QUALYS_BASE_URL = os.environ.get('QUALYS_BASE_URL') QUALYS_USERNAME = os.environ.get('QUALYS_USERNAME') QUALYS_PASSWORD = os.environ.get('QUALYS_PASSWORD') TRUNCATION_LIMIT = int(os.environ.get('TRUNCATION_LIMIT', '1000')) HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '300')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '50')) def _headers(): """Qualys FO API requires Basic auth and the X-Requested-With header.""" token = base64.b64encode(f'{QUALYS_USERNAME}:{QUALYS_PASSWORD}'.encode()).decode() return {'Authorization': f'Basic {token}', 'X-Requested-With': 'Cloud Run function'} def _extract_hosts(xml_text): """Parse the FO Host List XML into (list_of_host_dicts, next_page_url).""" doc = xmltodict.parse(xml_text) response = (doc.get('HOST_LIST_OUTPUT') or {}).get('RESPONSE') or {} host_list = (response.get('HOST_LIST') or {}).get('HOST') if host_list is None: hosts = [] elif isinstance(host_list, list): hosts = host_list else: hosts = [host_list] # When the result is truncated, RESPONSE contains <WARNING><URL>...</URL></WARNING>. warning = response.get('WARNING') next_url = warning.get('URL') if isinstance(warning, dict) else None return hosts, next_url def _write_hosts(bucket, hosts, page_num): """Write one batch of hosts to GCS as JSON, one host per line.""" ts_path = time.strftime('%Y/%m/%d', time.gmtime()) uniq = f'{int(time.time() * 1e6)}_{uuid.uuid4().hex[:8]}' key = f'{GCS_PREFIX}{ts_path}/qualys_asset_context_p{page_num:03d}_{uniq}.json' lines = [json.dumps(host, separators=(',', ':')) for host in hosts] bucket.blob(key).upload_from_string('\n'.join(lines), content_type='application/x-ndjson') return key @functions_framework.cloud_event def main(cloud_event): """Triggered by Pub/Sub to pull the Qualys host asset inventory and write it to GCS.""" if not all([GCS_BUCKET, QUALYS_BASE_URL, QUALYS_USERNAME, QUALYS_PASSWORD]): print('Error: missing required environment variables') return bucket = storage_client.bucket(GCS_BUCKET) session = requests.Session() session.headers.update(_headers()) url = f"{QUALYS_BASE_URL.rstrip('/')}/api/2.0/fo/asset/host/" params = { 'action': 'list', 'details': 'All', 'truncation_limit': TRUNCATION_LIMIT, # show_cloud_tags, host_metadata, and show_tags add the cloud provider # tags, cloud instance metadata, and asset tags that the parser maps. 'show_cloud_tags': 1, 'host_metadata': 'all', 'show_tags': 1, } pages = 0 total = 0 try: while pages < MAX_PAGES: resp = session.get(url, params=params, timeout=HTTP_TIMEOUT) resp.raise_for_status() hosts, next_url = _extract_hosts(resp.text) if hosts: _write_hosts(bucket, hosts, pages + 1) total += len(hosts) pages += 1 if not next_url: break # The pagination URL already includes every query parameter. url, params = next_url, None print(f'Wrote {total} hosts across {pages} pages') except Exception as e: print(f'Error collecting Qualys asset context: {e}') raiseCrie um arquivo
requirements.txtcom o seguinte conteúdo:functions-framework==3.* google-cloud-storage==2.* requests>=2.31.0 xmltodict>=0.13.0Clique em Implantar e aguarde a conclusão da implantação.
Criar um job do Cloud Scheduler
- Acesse o Cloud Scheduler.
- Clique em Criar job.
- Informe os seguintes detalhes de configuração:
- Nome: insira
qualys-asset-context-collector-daily. - Região: selecione a mesma região da função do Cloud Run.
- Frequência: insira
0 */6 * * *(a cada seis horas). - Fuso horário: selecione UTC.
- Tipo de destino: selecione Pub/Sub.
- Tema: selecione
qualys-asset-context-trigger. - Corpo da mensagem: insira
{}.
- Nome: insira
- Clique em Criar.
- Para verificar a configuração, clique em Executar forçadamente no job e confira os registros do Cloud Run e o bucket para confirmar se os registros de host foram gravados.
Configurar um feed no Google SecOps para ingerir registros de contexto de ativos da Qualys
- Acesse Configurações do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Na próxima página, clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Qualys asset context logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Contexto do recurso da Qualys como o Tipo de registro.
- Clique em Receber conta de serviço ao lado do campo Conta de serviço do Chronicle e copie o e-mail da conta de serviço.
- Clique em Próxima.
- Especifique valores para os seguintes parâmetros de entrada:
- URI do bucket de armazenamento: insira
gs://qualys-asset-context-logs/qualys-asset-context/. Esse URL precisa terminar com uma barra (/). - Opções de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência.
- Idade máxima do arquivo: inclui arquivos modificados nos últimos dias. O padrão é de 180 dias.
- URI do bucket de armazenamento: insira
- Clique em Próxima.
- Revise a nova configuração do feed na tela Finalizar e clique em Enviar.
Conceda à conta de serviço do Google SecOps acesso ao bucket
- 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: cole o e-mail da conta de serviço do Google SecOps da configuração do feed.
- Atribuir papéis: selecione Leitor de objetos do Storage ou Administrador de objetos do Storage se você escolheu uma opção de exclusão.
- Clique em Salvar.
Tabela de mapeamento de UDM
| Campo de registro | Mapeamento do UDM | Lógica |
|---|---|---|
ASSET_ID |
entity.asset.asset_id |
Mapeado diretamente |
DNS_DATA.HOSTNAME |
entity.asset.hostname |
Mapeado diretamente |
IP |
entity.asset.ip |
Mesclado |
OS |
entity.asset.platform_software.platform |
Mapeado: (?i)windows → WINDOWS |
ID |
entity.asset.product_object_id |
Mapeado diretamente |
dataset |
relations |
Mesclado |
| N/A | entity.asset.platform_software.platform |
Constante: WINDOWS |
| N/A | metadata.entity_type |
Constante: ASSET |
| N/A | metadata.product_name |
Constante: QUALYS ASSET CONTEXT |
| N/A | metadata.vendor_name |
Constante: QUALYS ASSET CONTEXT |
DNS_DATA.HOSTNAME |
entity.entity.asset.hostname |
Mapeado do registro de mudanças |
Registro de alterações
Ver o registro de alterações deste analisador
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.