Raccogliere i log di contesto degli asset Qualys

Supportato in:

Questo documento spiega come importare i log di contesto degli asset Qualys in Google Security Operations utilizzando Google Cloud Storage.

Il contesto degli asset Qualys fornisce dati di inventario degli asset host (indirizzi IP, nomi DNS, sistema operativo, metadati del provider cloud e tag) dalla piattaforma Qualys. Poiché questi dati vengono estratti dall'API Qualys, una funzione Cloud Run esegue periodicamente query sull'API Qualys Host List, converte la risposta XML in JSON e la scrive in un bucket Cloud Storage, dove viene importata da Google SecOps.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto cloud Google Cloud con le API Cloud Storage, Cloud Run, Pub/Sub e Cloud Scheduler abilitate
  • Autorizzazioni per creare e gestire bucket Cloud Storage, service account, funzioni Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Un account Qualys con accesso API (un utente con il ruolo Manager o un ruolo con accesso API e visibilità degli asset che vuoi raccogliere)

Crea un bucket Cloud Storage

  1. Vai alla console Google Cloud.
  2. Seleziona il tuo progetto o creane uno nuovo.
  3. Vai a Cloud Storage > Bucket.
  4. Fai clic su Crea bucket.
  5. Fornisci i seguenti dettagli di configurazione:
    • Assegna un nome al bucket: inserisci un nome univoco globale (ad esempio qualys-asset-context-logs).
    • Tipo di località: scegli in base alle tue esigenze (regione, a due regioni o multiregionale).
    • Località: seleziona la località (ad esempio, us-central1).
    • Classe di archiviazione: seleziona Standard.
    • Controllo dell'accesso: seleziona Uniforme.
  6. Fai clic su Crea.

Raccogliere le credenziali API Qualys

  1. Accedi all'applicazione web Qualys.
  2. Identifica o crea un utente API:
    • Vai a Utenti > Gestione utenti.
    • Utilizza un utente esistente o fai clic su Nuovo > Utente e assegna il ruolo Gestore (o un ruolo con accesso API).
  3. Registra il nome utente e la password dell'utente.

  4. Identifica l'URL del server API Qualys (è specifico per la tua piattaforma Qualys, ad esempio https://qualysapi.qg3.apps.qualys.com). Per trovarlo, consulta Identificare la piattaforma Qualys.

Crea un account di servizio per la funzione Cloud Run

  1. Vai a IAM e amministrazione > Service account.
  2. Fai clic su Crea service account.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome service account: inserisci qualys-asset-context-sa.
    • Descrizione del service account: inserisci Service account for Cloud Run function to collect Qualys asset context logs.
  4. Fai clic su Crea e continua.
  5. Concedi i seguenti ruoli:
    • Amministratore oggetti Storage: scrive i record host nel bucket.
    • Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione.
  6. Fai clic su Continua, quindi su Fine.

Concedi al account di servizio l'accesso al bucket Cloud Storage

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'email del account di servizio (qualys-asset-context-sa@<PROJECT_ID>.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea un argomento Pub/Sub

  1. Vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Nel campo ID argomento, inserisci qualys-asset-context-trigger.
  4. Fai clic su Crea.

Crea la funzione Cloud Run per raccogliere i log

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler. Chiama l'API Qualys FO Host List, converte la risposta XML in JSON e scrive i record host nel bucket.

  1. Vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione per utilizzare l'editor di codice incorporato.
  4. Fornisci i seguenti dettagli di configurazione:
    • Nome del servizio: inserisci qualys-asset-context-collector.
    • Regione: seleziona la regione corrispondente al tuo bucket (ad esempio us-central1).
    • Runtime: seleziona Python 3.12 o versioni successive.
  5. Nella sezione Trigger:
    1. Fai clic su Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, seleziona qualys-asset-context-trigger.
    4. Fai clic su Salva.
  6. Nella sezione Autenticazione, seleziona Richiedi autenticazione.
  7. Espandi Container, networking, sicurezza e nella scheda Sicurezza imposta Service account su qualys-asset-context-sa.

  8. Nella scheda Container > Variabili e secret, aggiungi le seguenti variabili di ambiente:

    Nome variabile Valore di esempio
    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. Imposta Timeout richiesta su 600 secondi e Memoria su 512 MiB o un valore superiore.

  10. Fai clic su Crea. Una volta creato il servizio, si apre l'editor di codice incorporato.

Aggiungi il codice per la funzione

  1. Nel campo Entry point della funzione, inserisci main.

  2. Nell'editor di codice incorporato, sostituisci i contenuti di main.py con il seguente codice:

    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. Crea un file requirements.txt con il seguente contenuto:

    functions-framework==3.*
google-cloud-storage==2.*
requests>=2.31.0
xmltodict>=0.13.0

  4. Fai clic su Esegui il deployment e attendi il completamento del deployment.

Crea un job Cloud Scheduler

  1. Vai a Cloud Scheduler.
  2. Fai clic su Crea job.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome: inserisci qualys-asset-context-collector-daily.
    • Regione: seleziona la stessa regione della funzione Cloud Run.
    • Frequenza: inserisci 0 */6 * * * (ogni sei ore).
    • Fuso orario: seleziona UTC.
    • Tipo di target: seleziona Pub/Sub.
    • Argomento: seleziona qualys-asset-context-trigger.
    • Corpo del messaggio: inserisci {}.
  4. Fai clic su Crea.
  5. Per verificare la configurazione, fai clic su Forza esecuzione nel job, quindi controlla i log di Cloud Run e il bucket per verificare che i record host siano stati scritti.

Configura un feed in Google SecOps per importare i log di contesto degli asset Qualys

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Nella pagina successiva, fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Qualys asset context logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Contesto asset Qualys come Tipo di log.
  7. Fai clic su Ottieni service account accanto al campo Service account Chronicle e copia l'email dell'account di servizio.
  8. Fai clic su Avanti.
  9. Specifica i valori per i seguenti parametri di input:
    • URI bucket di archiviazione: inserisci gs://qualys-asset-context-logs/qualys-asset-context/. Questo URL deve terminare con una barra finale (/).
    • Opzioni di eliminazione delle fonti: seleziona l'opzione di eliminazione in base alle tue preferenze.
    • Età massima file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.
  10. Fai clic su Avanti.
  11. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Concedi al account di servizio Google SecOps l'accesso al bucket

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'email del account di servizio Google SecOps dalla configurazione del feed.
    • Assegna i ruoli: seleziona Visualizzatore oggetti Storage (o Storage Object Admin se hai scelto un'opzione di eliminazione).
  6. Fai clic su Salva.

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
ASSET_ID entity.asset.asset_id Mappato direttamente
DNS_DATA.HOSTNAME entity.asset.hostname Mappato direttamente
IP entity.asset.ip Unita
OS entity.asset.platform_software.platform Mappato: (?i)windows → WINDOWS
ID entity.asset.product_object_id Mappato direttamente
dataset relations Unita
N/D entity.asset.platform_software.platform Costante: WINDOWS
N/D metadata.entity_type Costante: ASSET
N/D metadata.product_name Costante: QUALYS ASSET CONTEXT
N/D metadata.vendor_name Costante: QUALYS ASSET CONTEXT
DNS_DATA.HOSTNAME entity.entity.asset.hostname Mappato dal log delle modifiche

Log delle modifiche

Visualizza il log delle modifiche per questo parser

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.