Collecter les journaux du récepteur Netflow OpenTelemetry

Compatible avec :

Google SecOps SIEM

Remarque : Cette fonctionnalité est couverte par les Conditions des offres pré-DG des Conditions spécifiques du service Google Security Operations. Les fonctionnalités pré-DG (avant disponibilité générale) sont susceptibles de présenter une compatibilité limitée, et les modifications apportées à ces fonctionnalités peuvent ne pas être compatibles avec d'autres versions avant disponibilité générale. Pour en savoir plus, consultez les Instructions relatives aux services d'assistance technique Google SecOps et les Conditions spécifiques du service Google SecOps.
Si cette fonctionnalité n'est pas visible dans votre environnement, consultez Gérer les fonctionnalités en avant-première ou contactez votre administrateur système.

Ce document explique comment ingérer des journaux OpenTelemetry Netflow Receiver dans Google Security Operations à l'aide de Google Cloud Storage V2.

Le récepteur Netflow OpenTelemetry est un composant Open Source qui écoute le trafic UDP netflow, sflow et IPFIX, et le convertit en enregistrements de journaux OpenTelemetry. Cela permet de surveiller et d'analyser le trafic réseau, y compris l'identification des protocoles, l'analyse du volume de trafic, le suivi de l'utilisation des ports et les statistiques sur les octets/paquets.

Avant de commencer

Assurez-vous de remplir les conditions préalables suivantes :

• Une instance Google SecOps
• Un projet GCP pour lequel l'API Cloud Storage est activée
• Autorisations permettant de créer et de gérer des buckets GCS
• Autorisations permettant de gérer les stratégies IAM sur les buckets GCS
• Autorisations permettant de créer des services Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler
• Appareils réseau capables d'envoyer des données NetFlow, sFlow ou IPFIX
• Accès permettant de configurer les paramètres d'exportation des appareils réseau

Créer un bucket Google Cloud Storage

1. Accédez à la console Google Cloud.
2. Sélectionnez votre projet ou créez-en un.
3. Dans le menu de navigation, accédez à Cloud Storage > Buckets.
4. Cliquez sur Créer un bucket.

5. Fournissez les informations de configuration suivantes :

   Paramètre	Valeur
   Attribuez un nom au bucket	Saisissez un nom unique (par exemple, netflow-logs-bucket).
   Type d'emplacement	Choisissez celle qui correspond le mieux à vos besoins (régionale, birégionale ou multirégionale).
   Emplacement	Sélectionnez l'emplacement (par exemple, us-central1).
   Classe de stockage	Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
   Access control (Contrôle des accès)	Uniforme (recommandé)
   Outils de protection	Facultatif : Activez la gestion des versions d'objets ou la règle de conservation

6. Cliquez sur Créer.

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

La fonction Cloud Run a besoin d'un compte de service disposant des autorisations nécessaires pour écrire dans le bucket GCS et être appelée par Pub/Sub.

Créer un compte de service

1. Dans la console GCP, 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 netflow-collector-sa.
   • Description du compte de service : saisissez Service account for Cloud Run function to collect netflow logs.
4. Cliquez sur Créer et continuer.
5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
   1. Cliquez sur Sélectionner un rôle.
   2. Recherchez et sélectionnez Administrateur des objets Storage.
   3. Cliquez sur + Ajouter un autre rôle.
   4. Recherchez et sélectionnez Demandeur Cloud Run.
   5. Cliquez sur + Ajouter un autre rôle.
   6. Recherchez et sélectionnez Demandeur Cloud Functions.
6. Cliquez sur Continuer.
7. Cliquez sur OK.

Ces rôles sont requis pour :

• Administrateur des objets de l'espace de stockage : écrire des journaux dans le bucket GCS et gérer les fichiers d'état
• Demandeur Cloud Run : autorise Pub/Sub à appeler la fonction
• Demandeur Cloud Functions : autorise l'appel de fonctions

Accorder des autorisations IAM sur le bucket GCS

Accordez au compte de service des autorisations d'écriture dans le bucket GCS :

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom de votre bucket (par exemple, netflow-logs-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 (par exemple, netflow-collector-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

Créez un sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.

1. Dans la console GCP, accédez à Pub/Sub > Sujets.
2. Cliquez sur Créer un sujet.
3. Fournissez les informations de configuration suivantes :
   • ID du sujet : saisissez netflow-trigger.
   • Conservez les valeurs par défaut des autres paramètres.
4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux NetFlow

La fonction Cloud Run exécutera un collecteur OpenTelemetry qui reçoit les données NetFlow et les exporte vers GCS.

1. Dans la console GCP, accédez à Cloud Run.
2. Cliquez sur Créer un service.
3. Sélectionnez Fonction (utilisez un éditeur de modification directe pour créer une fonction).

4. Dans la section Configurer, fournissez les informations de configuration suivantes :

   Paramètre	Valeur
   Nom du service	netflow-collector
   Région	Sélectionnez la région correspondant à votre bucket GCS (par exemple, us-central1).
   Durée d'exécution	Sélectionnez Python 3.12 ou une version ultérieure.

5. Dans la section Déclencheur (facultatif) :

   1. Cliquez sur + Ajouter un déclencheur.
   2. Sélectionnez Cloud Pub/Sub.
   3. Dans Sélectionner un sujet Cloud Pub/Sub, choisissez le sujet Pub/Sub netflow-trigger.
   4. Cliquez sur Enregistrer.

6. Dans la section Authentification :

   1. Sélectionnez Exiger l'authentification.
   2. Consultez Identity and Access Management (IAM).
   Remarque : Pub/Sub gère automatiquement l'authentification lors de l'appel de la fonction.

7. Faites défiler la page vers le bas, puis développez Conteneurs, mise en réseau, sécurité.

8. Accédez à l'onglet Sécurité :

   • Compte de service : sélectionnez le compte de service netflow-collector-sa

9. Accédez à l'onglet Conteneurs :

   1. Cliquez sur Variables et secrets.
   2. Cliquez sur + Ajouter une variable pour chaque variable d'environnement :

   Nom de la variable	Exemple de valeur	Description
   GCS_BUCKET	netflow-logs-bucket	Nom du bucket GCS
   GCS_PREFIX	netflow	Préfixe des fichiers journaux
   NETFLOW_PORT	2055	Port du récepteur NetFlow
   NETFLOW_SCHEME	netflow	Type de Scheme : netflow, sflow ou ipfix
   NETFLOW_SOCKETS	4	Nombre de sockets UDP
   NETFLOW_WORKERS	8	Nombre de nœuds de calcul de décodeur

10. Dans la section Variables et secrets, faites défiler la page jusqu'à Requêtes :

    • Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).

11. Accédez à l'onglet Paramètres :

    • Dans la section Ressources :
      • Mémoire : sélectionnez 1 Gio ou plus.
      • CPU : sélectionnez 2.

12. Dans la section Scaling de révision :

    • Nombre minimal d'instances : saisissez 1.
    • Nombre maximal d'instances : saisissez 10.

13. Cliquez sur Créer.

14. Attendez que le service soit créé (une à deux minutes).

15. Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter un code de fonction

1. Saisissez main dans le champ Point d'entrée.

2. Dans l'éditeur de code intégré, créez trois fichiers :

   • main.py:

   import functions_framework
   from google.cloud import storage
   import json
   import os
   import subprocess
   import tempfile
   import signal
   import time
   from datetime import datetime, timezone
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   # Environment variables
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netflow')
   NETFLOW_PORT = os.environ.get('NETFLOW_PORT', '2055')
   NETFLOW_SCHEME = os.environ.get('NETFLOW_SCHEME', 'netflow')
   NETFLOW_SOCKETS = os.environ.get('NETFLOW_SOCKETS', '4')
   NETFLOW_WORKERS = os.environ.get('NETFLOW_WORKERS', '8')
   
   # Global process handle
   collector_process = None
   
   def create_collector_config():
     """Create OpenTelemetry Collector configuration."""
     timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
     file_path = f"/tmp/netflow_{timestamp}.ndjson"
   
     config = {
       'receivers': {
         'netflow': {
           'scheme': NETFLOW_SCHEME,
           'hostname': '0.0.0.0',
           'port': int(NETFLOW_PORT),
           'sockets': int(NETFLOW_SOCKETS),
           'workers': int(NETFLOW_WORKERS),
           'queue_size': 5000
         }
       },
       'processors': {
         'batch': {
           'timeout': '10s',
           'send_batch_size': 1000
         }
       },
       'exporters': {
         'file': {
           'path': file_path,
           'format': 'json'
         }
       },
       'service': {
         'pipelines': {
           'logs': {
             'receivers': ['netflow'],
             'processors': ['batch'],
             'exporters': ['file']
           }
         },
         'telemetry': {
           'logs': {
             'level': 'info'
           }
         }
       }
     }
   
     config_path = '/tmp/otel-config.yaml'
     with open(config_path, 'w') as f:
       import yaml
       yaml.dump(config, f)
   
     return config_path, file_path
   
   def upload_to_gcs(file_path):
     """Upload netflow logs to GCS."""
     if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
       print(f"No data to upload from {file_path}")
       return
   
     bucket = storage_client.bucket(GCS_BUCKET)
     timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
     object_key = f"{GCS_PREFIX}/netflow_{timestamp}.ndjson"
     blob = bucket.blob(object_key)
   
     blob.upload_from_filename(file_path, content_type='application/x-ndjson')
     print(f"Uploaded {os.path.getsize(file_path)} bytes to gs://{GCS_BUCKET}/{object_key}")
   
   def signal_handler(signum, frame):
     """Handle shutdown signals."""
     global collector_process
     if collector_process:
       print("Shutting down collector...")
       collector_process.terminate()
       collector_process.wait(timeout=10)
   
   @functions_framework.cloud_event
   def main(cloud_event):
     """
     Cloud Run function to run OpenTelemetry Collector for netflow collection.
   
     Args:
       cloud_event: CloudEvent object containing Pub/Sub message
     """
     global collector_process
   
     if not GCS_BUCKET:
       print('Error: GCS_BUCKET environment variable not set')
       return
   
     try:
       # Set up signal handlers
       signal.signal(signal.SIGTERM, signal_handler)
       signal.signal(signal.SIGINT, signal_handler)
   
       # Create collector configuration
       config_path, file_path = create_collector_config()
       print(f"Created collector config at {config_path}")
       print(f"Netflow receiver listening on {NETFLOW_SCHEME}://0.0.0.0:{NETFLOW_PORT}")
   
       # Start OpenTelemetry Collector
       collector_process = subprocess.Popen(
         ['/otelcol-contrib', '--config', config_path],
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         text=True
       )
   
       print(f"Started OpenTelemetry Collector (PID: {collector_process.pid})")
   
       # Run for collection period (e.g., 5 minutes)
       collection_time = 300
       print(f"Collecting netflow data for {collection_time} seconds...")
   
       start_time = time.time()
       while time.time() - start_time < collection_time:
         if collector_process.poll() is not None:
           stdout, stderr = collector_process.communicate()
           print(f"Collector exited unexpectedly")
           print(f"STDOUT: {stdout}")
           print(f"STDERR: {stderr}")
           break
         time.sleep(10)
   
       # Stop collector
       if collector_process.poll() is None:
         print("Stopping collector...")
         collector_process.terminate()
         collector_process.wait(timeout=10)
   
       # Upload collected data
       upload_to_gcs(file_path)
   
       # Cleanup
       if os.path.exists(file_path):
         os.remove(file_path)
       if os.path.exists(config_path):
         os.remove(config_path)
   
       print("Netflow collection completed successfully")
   
     except Exception as e:
       print(f'Error during netflow collection: {str(e)}')
       if collector_process and collector_process.poll() is None:
         collector_process.terminate()
       raise
   

   • requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   PyYAML==6.*
   

   • Dockerfile :

   FROM python:3.12-slim
   
   # Install OpenTelemetry Collector Contrib
   RUN apt-get update && apt-get install -y wget && \
     wget https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.144.0/otelcol-contrib_0.144.0_linux_amd64.deb && \
     dpkg -i otelcol-contrib_0.144.0_linux_amd64.deb && \
     rm otelcol-contrib_0.144.0_linux_amd64.deb && \
     apt-get clean
   
   # Set working directory
   WORKDIR /app
   
   # Copy requirements and install
   COPY requirements.txt .
   RUN pip install --no-cache-dir -r requirements.txt
   
   # Copy function code
   COPY main.py .
   
   # Expose netflow port
   EXPOSE 2055/udp
   
   # Run function
   CMD ["functions-framework", "--target=main", "--port=8080"]
   

3. Cliquez sur Déployer pour enregistrer et déployer la fonction.

4. Attendez la fin du déploiement (3 à 5 minutes).

Remarque : La fonction exécutera un collecteur OpenTelemetry qui écoute les données NetFlow et les exporte vers GCS.

Configurer les périphériques réseau pour envoyer des données NetFlow

Configurez vos appareils réseau (routeurs, commutateurs, pare-feu) pour exporter les données NetFlow, sFlow ou IPFIX vers la fonction Cloud Run.

Obtenir l'adresse IP externe de la fonction Cloud Run

1. Dans la console GCP, accédez à Cloud Run > Services.
2. Cliquez sur le nom de la fonction netflow-collector.
3. Copiez l'URL affichée en haut (par exemple, https://netflow-collector-xxxxx-uc.a.run.app).
4. Extrayez le nom d'hôte de l'URL.

5. Utilisez nslookup ou dig pour résoudre l'adresse IP :

   nslookup netflow-collector-xxxxx-uc.a.run.app
   

Remarque : Les services Cloud Run utilisent des adresses IP dynamiques. Pour les déploiements en production, envisagez d'utiliser un équilibreur de charge Cloud avec une adresse IP statique ou de déployer le collecteur sur Compute Engine.

• Exemple : configuration NetFlow d'un routeur Cisco

    ! Configure netflow exporter
    flow exporter OTEL-EXPORTER
    destination <CLOUD_RUN_IP>
    transport udp 2055
    source <INTERFACE>
    export-protocol netflow-v9
  
    ! Configure flow monitor
    flow monitor OTEL-MONITOR
    exporter OTEL-EXPORTER
    record netflow ipv4 original-input
  
    ! Apply to interface
    interface GigabitEthernet0/0
    ip flow monitor OTEL-MONITOR input
    ip flow monitor OTEL-MONITOR output
  

• Exemple : Configuration NetFlow générique

  Pour la plupart des périphériques réseau, configurez les paramètres suivants :

  • Version NetFlow : NetFlow v5, v9 ou IPFIX
  • IP du collecteur : adresse IP de la fonction Cloud Run
  • Port du collecteur : 2055 (ou port configuré)
  • Protocole : UDP
  • Délai avant expiration actif : 60 secondes (recommandé)
  • Délai d'inactivité : 15 secondes (recommandé)

Créer une tâche Cloud Scheduler

Cloud Scheduler publiera des messages dans le sujet Pub/Sub à intervalles réguliers, ce qui déclenchera la fonction Cloud Run.

1. Dans la console GCP, accédez à Cloud Scheduler.
2. Cliquez sur Créer une tâche.

3. Fournissez les informations de configuration suivantes :

   Paramètre	Valeur
   Nom	netflow-collector-hourly
   Région	Sélectionnez la même région que la fonction Cloud Run
   Fréquence	0 * * * * (toutes les heures, à l'heure pile)
   Fuseau horaire	Sélectionnez un fuseau horaire (UTC recommandé).
   Type de cible	Pub/Sub
   Sujet	Sélectionnez le sujet Pub/Sub netflow-trigger
   Corps du message	{} (objet JSON vide)

4. Cliquez sur Créer.

Options de fréquence de programmation

Choisissez la fréquence en fonction du volume de journaux et des exigences de latence :

Fréquence	Expression Cron	Cas d'utilisation
Toutes les 5 minutes	*/5 * * * *	Volume élevé, faible latence
Toutes les 15 minutes	*/15 * * * *	Volume moyen
Toutes les heures	0 * * * *	Standard (recommandé)
Toutes les 6 heures	0 */6 * * *	Traitement par lot à faible volume

Tester l'intégration

1. Dans la console Cloud Scheduler, recherchez le job netflow-collector-hourly.
2. Cliquez sur Exécuter de force pour déclencher le job manuellement.
3. Patientez pendant quelques secondes.
4. Accédez à Cloud Run > Services.
5. Cliquez sur le nom de la fonction netflow-collector.
6. Cliquez sur l'onglet Journaux.

7. Vérifiez que la fonction s'est exécutée correctement. Par exemple :

   Started OpenTelemetry Collector (PID: ...)
   Netflow receiver listening on netflow://0.0.0.0:2055
   Collecting netflow data for 300 seconds...
   Uploaded ... bytes to gs://netflow-logs-bucket/netflow/netflow_YYYYMMDD_HHMMSS.ndjson
   Netflow collection completed successfully
   

8. Accédez à Cloud Storage > Buckets.

9. Cliquez sur le nom du bucket netflow-logs-bucket.

10. Accédez au dossier de préfixe netflow/.

11. Vérifiez qu'un nouveau fichier .ndjson a été créé avec le code temporel actuel.

Si des erreurs s'affichent dans les journaux :

• Le collecteur s'est arrêté de manière inattendue : vérifiez la configuration du récepteur NetFlow
• Aucune donnée à importer : vérifiez que les périphériques réseau envoient les données NetFlow à l'adresse IP et au port corrects.
• Autorisation refusée : vérifiez que le compte de service dispose du rôle "Administrateur des objets de l'espace de stockage"
• Erreur de liaison de port : assurez-vous que le port 2055 n'est pas déjà utilisé.

Récupérer le compte de service Google SecOps

Google SecOps utilise un compte de service unique pour lire les données de votre bucket GCS. Vous devez accorder à ce compte de service l'accès à votre bucket.

Obtenir l'adresse e-mail du compte de service

1. Accédez à Paramètres SIEM> Flux.
2. Cliquez sur Ajouter un flux d'actualités.
3. Cliquez sur Configurer un flux unique.
4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, OpenTelemetry Netflow Logs).
5. Sélectionnez Google Cloud Storage V2 comme Type de source.
6. Sélectionnez NETFLOW_OTEL comme type de journal.

7. Cliquez sur Obtenir un compte de service. Une adresse e-mail de compte de service unique s'affiche. Exemple :

   [e-mail]
   

8. Copiez cette adresse e-mail pour l'utiliser à l'étape suivante.

   Remarque : Chaque instance Google SecOps possède un compte de service unique. N'utilisez pas de comptes de service provenant d'autres documentations ou exemples.

9. Cliquez sur Suivant.

10. Spécifiez les valeurs des paramètres d'entrée suivants :

    • URL du bucket de stockage : saisissez l'URI du bucket GCS avec le chemin d'accès au préfixe :

      gs://netflow-logs-bucket/netflow/
      

    Remarque : Incluez toujours la barre oblique finale (/) à la fin de l'URI.

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Jamais : ne supprime jamais aucun fichier après les transferts (recommandé pour les tests).
      • Supprimer les fichiers transférés : supprime les fichiers une fois le transfert réussi.

      • Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides une fois le transfert réussi.

        Remarque : Si vous sélectionnez une option de suppression, le compte de service doit disposer du rôle Administrateur des objets de l'espace de stockage au lieu du rôle Lecteur des objets de l'espace de stockage. Mettez à jour les autorisations IAM en conséquence.

    • Âge maximal du fichier : incluez les fichiers modifiés au cours des derniers jours (180 jours par défaut).

    • Espace de noms de l'élément : espace de noms de l'élément

    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux

11. Cliquez sur Suivant.

12. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps doit disposer du rôle Lecteur des objets Storage sur votre bucket GCS.

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom du bucket netflow-logs-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.
   • Attribuer des rôles : sélectionnez Lecteur d'objets Storage.
6. Cliquez sur Enregistrer.

Remarque : Si vous prévoyez d'utiliser l'option de suppression "Supprimer les fichiers transférés" ou "Supprimer les fichiers transférés et les répertoires vides", accordez le rôle Administrateur des objets Storage au lieu de Lecteur des objets Storage.

Table de mappage UDM

Le tableau suivant montre comment les champs de journaux du récepteur Netflow OpenTelemetry sont mappés aux champs UDM de Google SecOps :

Champ OpenTelemetry	Champ UDM	Description
source.address	principal.ip	Adresse IP source
source.port	principal.port	Numéro de port source
destination.address	target.ip	Adresse IP de destination
destination.port	target.port	Numéro du port de destination
network.transport	network.ip_protocol	Protocole de transport (tcp, udp)
network.type	network.ip_version	Version IP (IPv4, IPv6)
flow.io.bytes	network.sent_bytes	Nombre total d'octets transférés
flow.io.packets	network.sent_packets	Nombre total de paquets transférés
flow.type	metadata.product_log_id	Type de flux (NETFLOW_V5, NETFLOW_V9, SFLOW_5, IPFIX)
flow.start	metadata.event_timestamp	Heure de début du flux
flow.end	network.session_duration	Heure de fin du flux
flow.sampler_address	intermediary.ip	Adresse IP de l'exportateur NetFlow
flow.tcp_flags	network.tcp_flags	Options TCP
flow.io.bytes	event.idm.read_only_udm.network.sent_bytes	Mappé à partir du journal des modifications
source.port	event.idm.read_only_udm.principal.port	Mappé à partir du journal des modifications
destination.port	event.idm.read_only_udm.target.port	Mappé à partir du journal des modifications
network.transport	event.idm.read_only_udm.network.ip_protocol	Mappé à partir du journal des modifications
network.type	event.idm.read_only_udm.security_result.detection_fields	Mappé à partir du journal des modifications
flow.sequence_num	event.idm.read_only_udm.network.session_id	Mappé à partir du journal des modifications
flow.sampler_address	event.idm.read_only_udm.observer.ip	Mappé à partir du journal des modifications
flow.type	event.idm.read_only_udm.metadata.product_event_type	Mappé à partir du journal des modifications

Journal des modifications

Afficher le journal des modifications de cet analyseur

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