Recopila registros de contexto de activos de Qualys
En este documento, se explica cómo transferir registros de contexto de activos de Qualys a Google Security Operations con Google Cloud Storage.
El contexto de activos de Qualys proporciona datos de inventario de activos de host (direcciones IP, nombres de DNS, sistema operativo, metadatos de proveedores de servicios en la nube y etiquetas) desde la plataforma de Qualys. Dado que estos datos se extraen de la API de Qualys, una función de Cloud Run consulta periódicamente la API de Qualys Host List, convierte la respuesta XML en JSON y la escribe en un bucket de Cloud Storage, donde Google SecOps la ingiere.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos previos:
- Una instancia de Google SecOps
- Un proyecto de Google Cloud con las APIs de Cloud Storage, Cloud Run, Pub/Sub y Cloud Scheduler habilitadas
- Permisos para crear y administrar buckets de Cloud Storage, cuentas de servicio, funciones de Cloud Run, temas de Pub/Sub y trabajos de Cloud Scheduler
- Una cuenta de Qualys con acceso a la API (un usuario con el rol de Administrador o un rol que tenga acceso a la API y visibilidad de los activos que deseas recopilar)
Cree un bucket de Cloud Storage
- Ve a la consola de Google Cloud.
- Selecciona tu proyecto o crea uno nuevo.
- Ve a Cloud Storage > Buckets.
- Haz clic en Crear bucket.
- Proporciona los siguientes detalles de configuración:
- Asigna un nombre a tu bucket: Ingresa un nombre global único (por ejemplo,
qualys-asset-context-logs). - Tipo de ubicación: Elige según tus necesidades (región, birregional o multirregional).
- Ubicación: Selecciona la ubicación (por ejemplo,
us-central1). - Clase de almacenamiento: Selecciona Estándar.
- Control de acceso: Selecciona Uniforme.
- Asigna un nombre a tu bucket: Ingresa un nombre global único (por ejemplo,
- Haz clic en Crear.
Recopila credenciales de la API de Qualys
- Accede a la aplicación web de Qualys.
- Identifica o crea un usuario de la API:
- Ve a Usuarios > Administración de usuarios.
- Usa un usuario existente o haz clic en Nuevo > Usuario y asigna el rol de Administrador (o un rol con acceso a la API).
- Registra el nombre de usuario y la contraseña del usuario.
Identifica la URL de tu servidor de la API de Qualys (es específica para tu plataforma de Qualys, por ejemplo,
https://qualysapi.qg3.apps.qualys.com). Para encontrarla, consulta Cómo identificar tu plataforma de Qualys.
Crea una cuenta de servicio para la función de Cloud Run
- Ve a IAM y administración > Cuentas de servicio.
- Haga clic en Crear cuenta de servicio.
- Proporciona los siguientes detalles de configuración:
- Nombre de la cuenta de servicio: Ingresa
qualys-asset-context-sa. - Descripción de la cuenta de servicio: Ingresa
Service account for Cloud Run function to collect Qualys asset context logs.
- Nombre de la cuenta de servicio: Ingresa
- Haz clic en Crear y continuar.
- Otorga los siguientes roles:
- Administrador de objetos de Storage: Escribe registros de host en el bucket.
- Invocador de Cloud Run: Permite que Pub/Sub invoque la función.
- Haz clic en Continuar y, luego, en Listo.
Otorga acceso a la cuenta de servicio al bucket de Cloud Storage
- Ve a Cloud Storage > Buckets.
- Haz clic en el nombre de tu bucket.
- Ve a la pestaña Permisos.
- Haz clic en Otorgar acceso.
- Proporciona los siguientes detalles de configuración:
- Agregar entidades: Ingresa el correo electrónico de la cuenta de servicio (
qualys-asset-context-sa@<PROJECT_ID>.iam.gserviceaccount.com). - Asignar roles: Selecciona Administrador de objetos de Storage.
- Agregar entidades: Ingresa el correo electrónico de la cuenta de servicio (
- Haz clic en Guardar.
Crea un tema de Pub/Sub
- Ve a Pub/Sub > Temas.
- Haz clic en Crear un tema.
- En el campo ID de tema, ingresa
qualys-asset-context-trigger. - Haz clic en Crear.
Crea la función de Cloud Run para recopilar registros
La función de Cloud Run se activa con mensajes de Pub/Sub de Cloud Scheduler. Llama a la API de Qualys FO Host List, convierte la respuesta XML en JSON y escribe los registros de host en el bucket.
- Ve a Cloud Run.
- Haz clic en Crear servicio.
- Selecciona Función para usar el editor de código intercalado.
- Proporciona los siguientes detalles de configuración:
- Nombre del servicio: Ingresa
qualys-asset-context-collector. - Región: Selecciona la región que coincida con tu bucket (por ejemplo,
us-central1). - Entorno de ejecución: Selecciona Python 3.12 o una versión posterior.
- Nombre del servicio: Ingresa
- En la sección Activador, haz lo siguiente:
- Haz clic en Agregar activador.
- Selecciona Cloud Pub/Sub.
- En Selecciona un tema de Cloud Pub/Sub, selecciona
qualys-asset-context-trigger. - Haz clic en Guardar.
- En la sección Autenticación, selecciona Autenticación obligatoria.
- Expande Contenedores, redes y seguridad y, en la pestaña Seguridad, establece Cuenta de servicio en
qualys-asset-context-sa. En la pestaña Contenedores > Variables y secretos, agrega las siguientes variables de entorno:
Nombre de la variable Valor de ejemplo GCS_BUCKETqualys-asset-context-logsGCS_PREFIXqualys-asset-context/QUALYS_BASE_URLhttps://qualysapi.qg3.apps.qualys.comQUALYS_USERNAMEyour-qualys-usernameQUALYS_PASSWORDyour-qualys-passwordTRUNCATION_LIMIT1000HTTP_TIMEOUT300MAX_PAGES50Establece el Tiempo de espera de la solicitud en
600segundos y la Memoria en 512 MiB o más.Haz clic en Crear. Después de crear el servicio, se abre el editor de código intercalado.
Agrega el código de la función
- En el campo Punto de entrada de la función, ingresa
main. En el editor de código intercalado, reemplaza el contenido de
main.pypor el siguiente 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}') raiseCrea un archivo
requirements.txtcon el siguiente contenido:functions-framework==3.* google-cloud-storage==2.* requests>=2.31.0 xmltodict>=0.13.0Haz clic en Implementar y espera a que se complete la implementación.
Crea un trabajo de Cloud Scheduler
- Ve a Cloud Scheduler.
- Haz clic en Create job (Crear trabajo).
- Proporciona los siguientes detalles de configuración:
- Nombre: Ingresa
qualys-asset-context-collector-daily. - Región: Selecciona la misma región que la función de Cloud Run.
- Frecuencia: Ingresa
0 */6 * * *(cada seis horas). - Zona horaria: Selecciona UTC.
- Tipo de destino: Selecciona Pub/Sub.
- Tema: Selecciona
qualys-asset-context-trigger. - Cuerpo del mensaje: Ingresa
{}.
- Nombre: Ingresa
- Haz clic en Crear.
- Para verificar la configuración, haz clic en Forzar ejecución en el trabajo y, luego, revisa los registros de Cloud Run y el bucket para confirmar que se escribieron los registros del host.
Configura un feed en Google SecOps para transferir registros de contexto de recursos de Qualys
- Ve a Configuración de SIEM > Feeds.
- Haz clic en Add New Feed.
- En la siguiente página, haz clic en Configurar un solo feed.
- En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo,
Qualys asset context logs). - Selecciona Google Cloud Storage V2 como el Tipo de fuente.
- Selecciona Contexto del activo de Qualys como el Tipo de registro.
- Haz clic en Obtener cuenta de servicio junto al campo Cuenta de servicio de Chronicle y copia el correo electrónico de la cuenta de servicio.
- Haz clic en Siguiente.
- Especifica valores para los siguientes parámetros de entrada:
- URI del bucket de almacenamiento: Ingresa
gs://qualys-asset-context-logs/qualys-asset-context/. Esta URL debe terminar con una barra diagonal final (/). - Opciones de eliminación del código fuente: Selecciona la opción de eliminación según tu preferencia.
- Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es de 180 días.
- URI del bucket de almacenamiento: Ingresa
- Haz clic en Siguiente.
- Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.
Otorga acceso a la cuenta de servicio de Google SecOps al bucket
- Ve a Cloud Storage > Buckets.
- Haz clic en el nombre de tu bucket.
- Ve a la pestaña Permisos.
- Haz clic en Otorgar acceso.
- Proporciona los siguientes detalles de configuración:
- Agregar entidades: Pega el correo electrónico de la cuenta de servicio de Google SecOps de la configuración del feed.
- Asignar roles: Selecciona Visualizador de objetos de Storage (o Administrador de objetos de Storage si elegiste una opción de eliminación).
- Haz clic en Guardar.
Tabla de asignación de UDM
| Campo de registro | Asignación de UDM | Lógica |
|---|---|---|
ASSET_ID |
entity.asset.asset_id |
Asignada directamente |
DNS_DATA.HOSTNAME |
entity.asset.hostname |
Asignada directamente |
IP |
entity.asset.ip |
Combinado |
OS |
entity.asset.platform_software.platform |
Asignado: (?i)windows → WINDOWS |
ID |
entity.asset.product_object_id |
Asignada directamente |
dataset |
relations |
Combinado |
| 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 |
Se asignó desde el registro de cambios |
Registro de cambios
Consulta el registro de cambios de este analizador
¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.