Collecter les journaux de contexte des composants Qualys

Ce document explique comment ingérer des journaux de contexte d'assets Qualys dans Google Security Operations à l'aide de Google Cloud Storage.

Le contexte des composants Qualys fournit des données d'inventaire des composants hôtes (adresses IP, noms DNS, système d'exploitation, métadonnées du fournisseur de services cloud et tags) à partir de la plate-forme Qualys. Étant donné que ces données sont extraites de l'API Qualys, une fonction Cloud Run interroge régulièrement l'API Qualys Host List, convertit la réponse XML en JSON et l'écrit dans un bucket Cloud Storage, où Google SecOps l'ingère.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

• Une instance Google SecOps
• Un projet Google Cloud avec les API Cloud Storage, Cloud Run, Pub/Sub et Cloud Scheduler activées
• Autorisations permettant de créer et de gérer des buckets Cloud Storage, des comptes de service, des fonctions Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler
• Un compte Qualys avec accès à l'API (un utilisateur avec le rôle Manager ou un rôle qui a accès à l'API et qui peut voir les composants que vous souhaitez collecter)

Créer un bucket Cloud Storage

1. Accédez à Google Cloud Console.
2. Sélectionnez votre projet ou créez-en un.
3. Accédez à Cloud Storage > Buckets.
4. Cliquez sur Créer un bucket.
5. Fournissez les informations de configuration suivantes :
   • Attribuer un nom au bucket : saisissez un nom unique (par exemple, qualys-asset-context-logs).
   • Type d'emplacement : choisissez en fonction de vos besoins (régional, birégional ou multirégional).
   • Emplacement : sélectionnez l'emplacement (par exemple, us-central1).
   • Classe de stockage : sélectionnez Standard.
   • Contrôle des accès : sélectionnez Uniforme.
6. Cliquez sur Créer.

Collecter les identifiants de l'API Qualys

1. Connectez-vous à l'application Web Qualys.
2. Identifiez ou créez un utilisateur de l'API :
   • Accédez à Utilisateurs > Gestion des utilisateurs.
   • Utilisez un utilisateur existant ou cliquez sur Nouveau > Utilisateur, puis attribuez le rôle Administrateur (ou un rôle avec accès à l'API).
3. Notez le nom d'utilisateur et le mot de passe de l'utilisateur.

4. Identifiez l'URL de votre serveur d'API Qualys (elle est spécifique à votre plate-forme Qualys, par exemple https://qualysapi.qg3.apps.qualys.com). Pour la trouver, consultez Identifier votre plate-forme Qualys.

Créer un compte de service pour la fonction Cloud Run

1. Accédez à IAM et administration > Comptes de service.
2. Cliquez sur Créer un compte de service.
3. Fournissez les informations de configuration suivantes :
   • Nom du compte de service : saisissez qualys-asset-context-sa.
   • Description du compte de service : saisissez Service account for Cloud Run function to collect Qualys asset context logs.
4. Cliquez sur Créer et continuer.
5. Attribuez les rôles suivants :
   • Administrateur des objets Storage : écrire des enregistrements d'hôte dans le bucket.
   • Demandeur Cloud Run : permet à Pub/Sub d'appeler la fonction.
6. Cliquez sur Continue (Continuer), puis sur Done (OK).

Accorder au compte de service l'accès au bucket Cloud Storage

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom de votre bucket.
3. Accédez à l'onglet Autorisations.
4. Cliquez sur Accorder l'accès.
5. Fournissez les informations de configuration suivantes :
   • Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (qualys-asset-context-sa@<PROJECT_ID>.iam.gserviceaccount.com).
   • Attribuer des rôles : sélectionnez Administrateur des objets Storage.
6. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

1. Accédez à Pub/Sub > Sujets.
2. Cliquez sur Create topic (Créer un sujet).
3. Dans le champ ID du thème, saisissez qualys-asset-context-trigger.
4. Cliquez sur Créer.

Créer la fonction Cloud Run pour collecter les journaux

La fonction Cloud Run est déclenchée par des messages Pub/Sub provenant de Cloud Scheduler. Il appelle l'API Qualys FO Host List, convertit la réponse XML en JSON et écrit les enregistrements d'hôte dans le bucket.

1. Accédez à Cloud Run.
2. Cliquez sur Créer un service.
3. Sélectionnez Fonction pour utiliser l'éditeur de code intégré.
4. Fournissez les informations de configuration suivantes :
   • Nom du service : saisissez qualys-asset-context-collector.
   • Région : sélectionnez la région correspondant à votre bucket (par exemple, us-central1).
   • Environnement d'exécution : sélectionnez Python 3.12 ou version ultérieure.
5. Dans la section Déclencheur :
   1. Cliquez sur Ajouter un déclencheur.
   2. Sélectionnez Cloud Pub/Sub.
   3. Dans Sélectionner un sujet Cloud Pub/Sub, sélectionnez qualys-asset-context-trigger.
   4. Cliquez sur Enregistrer.
6. Dans la section Authentification, sélectionnez Exiger l'authentification.
7. Développez Conteneurs, mise en réseau, sécurité, puis dans l'onglet Sécurité, définissez Compte de service sur qualys-asset-context-sa.

8. Dans l'onglet Conteneurs> Variables et secrets, ajoutez les variables d'environnement suivantes :

   Nom de la variable	Exemple de valeur
   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. Définissez Délai avant expiration de la requête sur 600 secondes et Mémoire sur 512 Mio ou plus.

10. Cliquez sur Créer. Une fois le service créé, l'éditeur de code intégré s'ouvre.

Ajouter le code de la fonction

1. Dans le champ Point d'entrée de la fonction, saisissez main.

2. Dans l'éditeur de code intégré, remplacez le contenu de main.py par le code suivant :

   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. Créez un fichier requirements.txt avec le contenu suivant :

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

4. Cliquez sur Déployer et attendez la fin du déploiement.

Créer une tâche Cloud Scheduler

1. Accédez à Cloud Scheduler.
2. Cliquez sur Créer une tâche.
3. Fournissez les informations de configuration suivantes :
   • Nom : saisissez qualys-asset-context-collector-daily.
   • Région : sélectionnez la même région que la fonction Cloud Run.
   • Fréquence : saisissez 0 */6 * * * (toutes les six heures).
   • Fuseau horaire : sélectionnez UTC.
   • Type de cible : sélectionnez Pub/Sub.
   • Sujet : sélectionnez qualys-asset-context-trigger.
   • Corps du message : saisissez {}.
4. Cliquez sur Créer.
5. Pour vérifier la configuration, cliquez sur Exécuter de force sur le job, puis consultez les journaux Cloud Run et le bucket pour confirmer que les enregistrements d'hôte ont été écrits.

Configurer un flux dans Google SecOps pour ingérer les journaux de contexte des composants Qualys

1. Accédez à Paramètres SIEM> Flux.
2. Cliquez sur Add New Feed (Ajouter un flux).
3. Sur la page suivante, cliquez sur Configurer un seul flux.
4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Qualys asset context logs).
5. Sélectionnez Google Cloud Storage V2 comme Type de source.
6. Sélectionnez Contexte des assets Qualys comme Type de journal.
7. Cliquez sur Obtenir le compte de service à côté du champ Compte de service Chronicle, puis copiez l'adresse e-mail du compte de service.
8. Cliquez sur Suivant.
9. Spécifiez les valeurs des paramètres d'entrée suivants :
   • URI du bucket de stockage : saisissez gs://qualys-asset-context-logs/qualys-asset-context/. Cette URL doit se terminer par une barre oblique (/).
   • Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
   • Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours. La valeur par défaut est de 180 jours.
10. Cliquez sur Suivant.
11. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Accorder au compte de service Google SecOps l'accès au bucket

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom de votre bucket.
3. Accédez à l'onglet Autorisations.
4. Cliquez sur Accorder l'accès.
5. Fournissez les informations de configuration suivantes :
   • Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps à partir de la configuration du flux.
   • Attribuer des rôles : sélectionnez Lecteur d'objets Storage (ou Administrateur des objets Storage si vous avez choisi une option de suppression).
6. Cliquez sur Enregistrer.

Table de mappage UDM

Journal des modifications

Afficher le journal des modifications pour ce parseur

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.