Coletar registros de contexto de recursos da Qualys

Compatível com:

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

  1. Acesse o Console do Google Cloud.
  2. Selecione ou crie um projeto.
  3. Acesse Cloud Storage > Buckets.
  4. Clique em Criar bucket.
  5. 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.
  6. Clique em Criar.

Coletar credenciais da API Qualys

  1. Faça login no aplicativo da Web Qualys.
  2. 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).
  3. Registre o nome de usuário e a senha do usuário.
  4. 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

  1. Acesse IAM e administrador > Contas de serviço.
  2. Clique em Criar conta de serviço.
  3. 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.
  4. Clique em Criar e continuar.
  5. 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.
  6. Clique em Continuar e depois em Concluído.

Conceder à conta de serviço acesso ao bucket do Cloud Storage

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. 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.
  6. Clique em Salvar.

Criar um tópico do Pub/Sub

  1. Acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.
  3. No campo ID do tópico, insira qualys-asset-context-trigger.
  4. 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.

  1. Acesse o Cloud Run.
  2. Clique em Criar serviço.
  3. Selecione Função para usar o editor de código in-line.
  4. 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.
  5. Na seção Acionador:
    1. Clique em Adicionar acionador.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, selecione qualys-asset-context-trigger.
    4. Clique em Salvar.
  6. Na seção Autenticação, selecione Exigir autenticação.
  7. Abra Contêineres, rede, segurança e, na guia Segurança, defina Conta de serviço como qualys-asset-context-sa.
  8. Na guia Contêineres > Variáveis e secrets, adicione as seguintes variáveis de ambiente:

    Nome da variável Valor de exemplo
    GCS_BUCKET qualys-asset-context-logs
    GCS_PREFIX qualys-asset-context/
    QUALYS_BASE_URL https://qualysapi.qg3.apps.qualys.com
    QUALYS_USERNAME your-qualys-username
    QUALYS_PASSWORD your-qualys-password
    TRUNCATION_LIMIT 1000
    HTTP_TIMEOUT 300
    MAX_PAGES 50
  9. Defina o Tempo limite da solicitação como 600 segundos e a Memória como 512 MiB ou mais.

  10. Clique em Criar. Depois que o serviço é criado, o editor de código inline é aberto.

Adicionar o código da função

  1. No campo Ponto de entrada da função, insira main.
  2. No editor de código inline, substitua o conteúdo de main.py pelo 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}')
            raise
    
  3. Crie um arquivo requirements.txt com o seguinte conteúdo:

    functions-framework==3.*
    google-cloud-storage==2.*
    requests>=2.31.0
    xmltodict>=0.13.0
    
  4. Clique em Implantar e aguarde a conclusão da implantação.

Criar um job do Cloud Scheduler

  1. Acesse o Cloud Scheduler.
  2. Clique em Criar job.
  3. 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 {}.
  4. Clique em Criar.
  5. 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

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em Adicionar novo feed.
  3. Na próxima página, clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Qualys asset context logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione Contexto do recurso da Qualys como o Tipo de registro.
  7. 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.
  8. Clique em Próxima.
  9. 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.
  10. Clique em Próxima.
  11. 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

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. 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.
  6. 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.