Collecter les journaux de surveillance continue Qualys
Ce code d'analyseur Logstash extrait d'abord des champs tels que l'adresse IP source, l'utilisateur, la méthode et le protocole d'application à partir de messages de journaux bruts à l'aide de modèles Grok. Il mappe ensuite des champs spécifiques des données de journal brutes à leurs champs correspondants dans le modèle de données unifié (UDM), effectue des conversions de types de données et enrichit les données avec des libellés et des métadonnées supplémentaires avant de structurer la sortie au format UDM souhaité.
Avant de commencer
Assurez-vous de remplir les conditions préalables suivantes :
- Instance Google Security Operations.
- Accès privilégié à Google Cloud.
- Accès privilégié à Qualys.
Activez les API requises :
- Connectez-vous à la console Google Cloud .
- Accédez à API et services > Bibliothèque.
- Recherchez les API suivantes et activez-les :
- API Cloud Functions
- API Cloud Scheduler
- Cloud Pub/Sub (requis pour que Cloud Scheduler appelle des fonctions)
Créer un bucket de stockage Google Cloud
- Connectez-vous à la console Google Cloud .
Accédez à la page Buckets Cloud Storage.
Cliquez sur Créer.
Configurez le bucket :
- Nom : saisissez un nom unique qui répond aux exigences de dénomination des buckets (par exemple, qualys-asset-bucket).
- Choisissez où stocker vos données : sélectionnez un emplacement.
- Choisir une classe de stockage pour vos données : sélectionnez une classe de stockage par défaut pour le bucket, ou bien classe automatique pour une gestion automatique des classes de stockage.
- Choisir comment contrôler l'accès aux objets : sélectionnez non pour appliquer la protection contre l'accès public, puis sélectionnez un modèle de contrôle des accès pour les objets de votre bucket.
- Classe de stockage : choisissez en fonction de vos besoins (par exemple, Standard).
Cliquez sur Créer.
Créer un compte de service Google Cloud
- Connectez-vous à la console Google Cloud .
- Accédez à IAM et administration > Comptes de service.
- Créez un compte de service.
- Attribuez-lui un nom descriptif (par exemple, qualys-user).
- Accordez au compte de service le rôle Administrateur des objets de l'espace de stockage sur le bucket GCS que vous avez créé à l'étape précédente.
- Attribuez le rôle Demandeur Cloud Functions au compte de service.
- Créez une clé SSH pour le compte de service.
- Téléchargez un fichier de clé JSON pour le compte de service. Conservez ce fichier dans un endroit sécurisé.
Facultatif : Créez un utilisateur API dédié dans Qualys
- Connectez-vous à la console Qualys.
- Accédez à Utilisateurs.
- Cliquez sur Nouveau > Utilisateur.
- Saisissez les informations générales requises pour l'utilisateur.
- Sélectionnez l'onglet Rôle utilisateur.
- Assurez-vous que la case Accès API est cochée pour le rôle.
- Cliquez sur Enregistrer.
Identifier votre URL d'API Qualys spécifique
Option 1
Identifiez vos URL comme indiqué dans la section Identification de la plate-forme.
Option 2
- Connectez-vous à la console Qualys.
- Accédez à Aide > À propos.
- Faites défiler la page pour afficher ces informations sous "Centre des opérations de sécurité (SOC)".
- Copiez l'URL de l'API Qualys.
Configurer la fonction Cloud Functions
- Accédez à Cloud Functions dans la console Google Cloud .
- Cliquez sur Créer une fonction.
Configurez la fonction :
- Nom : saisissez un nom pour votre fonction (par exemple, fetch-qualys-cm-alerts).
- Région : sélectionnez une région proche de votre bucket.
- Environnement d'exécution : Python 3.10 (ou l'environnement d'exécution de votre choix).
- Déclencheur : choisissez le déclencheur HTTP si nécessaire ou Cloud Pub/Sub pour une exécution planifiée.
- Authentification : sécurisé avec authentification.
- Écrivez le code avec un éditeur intégré :
```python from google.cloud import storage import requests import base64 import json # Google Cloud Storage Configuration BUCKET_NAME = "<bucket-name>" FILE_NAME = "qualys_cm_alerts.json" # Qualys API Credentials QUALYS_USERNAME = "<qualys-username>" QUALYS_PASSWORD = "<qualys-password>" QUALYS_BASE_URL = "https://<qualys_base_url>" def fetch_cm_alerts(): """Fetch alerts from Qualys Continuous Monitoring.""" auth = base64.b64encode(f"{QUALYS_USERNAME}:{QUALYS_PASSWORD}".encode()).decode() headers = { "Authorization": f"Basic {auth}", "Content-Type": "application/xml" } payload = """ <ServiceRequest> <filters> <Criteria field="alert.date" operator="GREATER">2024-01-01</Criteria> </filters> </ServiceRequest> """ response = requests.post(f"{QUALYS_BASE_URL}/qps/rest/2.0/search/cm/alert", headers=headers, data=payload) response.raise_for_status() return response.json() def upload_to_gcs(data): """Upload data to Google Cloud Storage.""" client = storage.Client() bucket = client.get_bucket(BUCKET_NAME) blob = bucket.blob(FILE_NAME) blob.upload_from_string(json.dumps(data, indent=2), content_type="application/json") def main(request): """Cloud Function entry point.""" try: alerts = fetch_cm_alerts() upload_to_gcs(alerts) return "Qualys CM alerts uploaded to Cloud Storage successfully!" except Exception as e: return f"An error occurred: {e}", 500 ```
Une fois la configuration terminée, cliquez sur Déployer.
Configurer Cloud Scheduler
- Accédez à Cloud Scheduler dans la console Google Cloud .
- Cliquez sur Créer une tâche.
Configurez le job :
- Nom : saisissez le nom de votre job (par exemple, trigger-fetch-qualys-cm-alerts).
- Fréquence : utilisez la syntaxe cron pour spécifier la planification (par exemple,
0 * * * *
pour exécuter la tâche toutes les heures). - Fuseau horaire : définissez votre fuseau horaire préféré.
- Type de déclencheur : sélectionnez HTTP.
- URL du déclencheur : saisissez l'URL de la fonction Cloud (disponible dans les détails de la fonction après le déploiement).
- Méthode : sélectionnez POST.
Créez le job.
Configurer des flux
Il existe deux points d'entrée différents pour configurer les flux dans la plate-forme Google SecOps :
- Paramètres SIEM> Flux
- Plate-forme de contenu > Packs de contenu
Configurer des flux à partir de Paramètres SIEM > Flux
Pour configurer un flux, procédez comme suit :
- Accédez à Paramètres SIEM > Flux.
- Cliquez sur Add New Feed (Ajouter un flux).
- Sur la page suivante, cliquez sur Configurer un seul flux.
- Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Journaux de surveillance continue Qualys).
- Sélectionnez Google Cloud Storage comme Type de source.
- Sélectionnez Surveillance continue Qualys comme Type de journal.
- Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
- URI du bucket Storage : URI source du bucket Storage Google Cloud .
- L'URI est un : sélectionnez Fichier unique.
- Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
Cliquez sur Suivant.
Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.
Configurer des flux depuis le Hub de contenu
Indiquez les valeurs des champs suivants :
- URI du bucket Storage : URI source du bucket Storage Google Cloud .
- L'URI est un : sélectionnez Fichier unique.
- Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
Options avancées
- Nom du flux : valeur préremplie qui identifie le flux.
- Type de source : méthode utilisée pour collecter les journaux dans Google SecOps.
- Espace de noms de l'élément : espace de noms associé au flux.
- Libellés d'ingestion : libellés appliqués à tous les événements de ce flux.
Table de mappage UDM
Champ de journal | Mappage UDM | Logique |
---|---|---|
Alert.alertInfo.appVersion | metadata.product_version | Mappé directement à partir de Alert.alertInfo.appVersion |
Alert.alertInfo.operatingSystem | principal.platform_version | Mappé directement à partir de Alert.alertInfo.operatingSystem |
Alert.alertInfo.port | additional.fields.value.string_value | Mappé directement à partir de Alert.alertInfo.port et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Alert port" |
Alert.alertInfo.protocol | network.ip_protocol | Mappé directement à partir de Alert.alertInfo.protocol |
Alert.alertInfo.sslIssuer | network.tls.client.certificate.issuer | Mappé directement à partir de Alert.alertInfo.sslIssuer |
Alert.alertInfo.sslName | additional.fields.value.string_value | Directement mappé à partir de Alert.alertInfo.sslName et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Nom SSL" |
Alert.alertInfo.sslOrg | additional.fields.value.string_value | Directement mappé à partir de Alert.alertInfo.sslOrg et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "SSL Org" |
Alert.alertInfo.ticketId | additional.fields.value.string_value | Mappé directement à partir de Alert.alertInfo.ticketId et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Ticket Id" |
Alert.alertInfo.vpeConfidence | additional.fields.value.string_value | Directement mappé à partir de Alert.alertInfo.vpeConfidence et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "VPE Confidence" |
Alert.alertInfo.vpeStatus | additional.fields.value.string_value | Directement mappé à partir de Alert.alertInfo.vpeStatus et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "VPE Confidence" |
Alert.eventType | additional.fields.value.string_value | Directement mappé à partir de Alert.eventType et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Event Type" (Type d'événement) |
Alert.hostname | principal.hostname | Mappé directement à partir de Alert.hostname |
Alert.id | security_result.threat_id | Mappé directement à partir de Alert.id |
Alert.ipAddress | principal.ip | Mappé directement à partir de Alert.ipAddress |
Alert.profile.id | additional.fields.value.string_value | Directement mappé à partir de Alert.profile.id et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "ID de profil" |
Alert.profile.title | additional.fields.value.string_value | Directement mappé à partir de Alert.profile.title et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Profile Title" |
Alert.qid | vulnerability.name | Mappé en tant que "QID : Alert.qid |
Alert.source | additional.fields.value.string_value | Directement mappé à partir de Alert.source et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Alert Source" |
Alert.triggerUuid | metadata.product_log_id | Mappé directement à partir de Alert.triggerUuid |
Alert.vulnCategory | additional.fields.value.string_value | Directement mappé à partir de Alert.vulnCategory et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Vulnerability Category" (Catégorie de failles) |
Alert.vulnSeverity | vulnerability.severity | Mappé en fonction de la valeur de Alert.vulnSeverity : 1 à 3 : FAIBLE, 4 à 6 : MOYEN, 7 à 8 : ÉLEVÉ |
Alert.vulnTitle | vulnerability.description | Mappé directement à partir de Alert.vulnTitle |
Alert.vulnType | additional.fields.value.string_value | Directement mappé à partir de Alert.vulnType et ajouté en tant que paire clé-valeur dans additional.fields avec la clé "Vulnerability Type" |
Hôte | principal.ip | Analysé à partir de la ligne de journal "Host: |
edr.client.ip_addresses | Copiés depuis l'appareil principal.ip |
|
edr.client.hostname | Copiés depuis l'appareil principal.hostname |
|
edr.raw_event_name | Définissez sur "STATUS_UPDATE" si Alert.ipAddress , Alert.hostname ou src_ip sont présents, sinon définissez sur "GENERIC_EVENT". |
|
metadata.event_timestamp | Extrait des champs Alert.eventDate ou timestamp . Alert.eventDate est prioritaire s'il existe. Sinon, timestamp est utilisé. L'horodatage est converti au format UTC. |
|
metadata.event_type | Même logique que pour edr.raw_event_name |
|
metadata.log_type | Définissez-le sur "QUALYS_CONTINUOUS_MONITORING". | |
metadata.product_name | Définissez-le sur "QUALYS_CONTINUOUS_MONITORING". | |
metadata.vendor_name | Définissez-le sur "QUALYS_CONTINUOUS_MONITORING". | |
network.application_protocol | Analysé à partir de la ligne de journal " |
|
network.http.method | Analyse de la ligne de journal " |
|
timestamp | event.timestamp | Extrait des champs Alert.eventDate ou timestamp . Alert.eventDate est prioritaire s'il existe. Sinon, timestamp est utilisé. L'horodatage est converti au format UTC. |
Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.