Raccogli i log del ricevitore Netflow di OpenTelemetry

Supportato in:

Google SecOps SIEM

Nota: questa funzionalità è coperta dai Termini delle Offerte pre-GA dei Termini di servizio specifici di Google Security Operations. Le funzionalità pre-GA potrebbero avere supporto limitato e le modifiche alle funzionalità pre-GA potrebbero non essere compatibili con altre versioni pre-GA. Per saperne di più, consulta le linee guida per i servizi di assistenza tecnica di Google SecOps e i Termini di servizio specifici di Google SecOps.
Se questa funzionalità non è visibile nel tuo ambiente, consulta Gestire le funzionalità di anteprima o contatta l'amministratore di sistema.

Questo documento spiega come importare i log del ricevitore OpenTelemetry Netflow in Google Security Operations utilizzando Google Cloud Storage V2.

Il ricevitore OpenTelemetry Netflow è un componente open source che ascolta il traffico UDP netflow, sflow e IPFIX e lo converte in record di log OpenTelemetry. Ciò consente il monitoraggio e l'analisi del traffico di rete, inclusi l'identificazione del protocollo, l'analisi del volume di traffico, il monitoraggio dell'utilizzo delle porte e le statistiche su byte/pacchetti.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

• Un'istanza Google SecOps
• Un progetto GCP con l'API Cloud Storage abilitata
• Autorizzazioni per creare e gestire bucket GCS
• Autorizzazioni per gestire le policy IAM nei bucket GCS
• Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
• Dispositivi di rete in grado di inviare dati NetFlow, sFlow o IPFIX
• Accesso per configurare le impostazioni di esportazione dei dispositivi di rete

Crea un bucket Google Cloud Storage

1. Vai alla console Google Cloud.
2. Seleziona il tuo progetto o creane uno nuovo.
3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
4. Fai clic su Crea bucket.

5. Fornisci i seguenti dettagli di configurazione:

   Impostazione	Valore
   Assegna un nome al bucket	Inserisci un nome univoco globale (ad esempio netflow-logs-bucket).
   Tipo di località	Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
   Località	Seleziona la posizione (ad esempio, us-central1).
   Classe di archiviazione	Standard (consigliato per i log a cui si accede di frequente)
   Controllo dell'accesso	Uniforme (consigliato)
   Strumenti di protezione	(Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

6. Fai clic su Crea.

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni per scrivere nel bucket GCS ed essere richiamata da Pub/Sub.

Crea service account

1. Nella console Google Cloud, vai a IAM e amministrazione > Account di servizio.
2. Fai clic su Crea account di servizio.
3. Fornisci i seguenti dettagli di configurazione:
   • Nome del service account: inserisci netflow-collector-sa
   • Descrizione service account: inserisci Service account for Cloud Run function to collect netflow logs
4. Fai clic su Crea e continua.
5. Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
   1. Fai clic su Seleziona un ruolo.
   2. Cerca e seleziona Storage Object Admin.
   3. Fai clic su + Aggiungi un altro ruolo.
   4. Cerca e seleziona Cloud Run Invoker.
   5. Fai clic su + Aggiungi un altro ruolo.
   6. Cerca e seleziona Cloud Functions Invoker.
6. Fai clic su Continua.
7. Fai clic su Fine.

Questi ruoli sono necessari per:

• Storage Object Admin: scrive i log nel bucket GCS e gestisce i file di stato
• Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
• Cloud Functions Invoker: consente la chiamata di funzioni

Concedi le autorizzazioni IAM sul bucket GCS

Concedi al account di servizio le autorizzazioni in scrittura sul bucket GCS:

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

Crea argomento Pub/Sub

Crea un argomento Pub/Sub a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

1. Nella console Google Cloud, vai a Pub/Sub > Argomenti.
2. Fai clic su Crea argomento.
3. Fornisci i seguenti dettagli di configurazione:
   • ID argomento: inserisci netflow-trigger
   • Lascia invariate le altre impostazioni predefinite.
4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log NetFlow

La funzione Cloud Run eseguirà un raccoglitore OpenTelemetry che riceve i dati NetFlow e li esporta in GCS.

1. Nella console Google Cloud, vai a Cloud Run.
2. Fai clic su Crea servizio.
3. Seleziona Funzione (utilizza un editor in linea per creare una funzione).

4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

   Impostazione	Valore
   Nome servizio	netflow-collector
   Regione	Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)
   Tempo di esecuzione	Seleziona Python 3.12 o versioni successive.

5. Nella sezione Trigger (facoltativo):

   1. Fai clic su + Aggiungi trigger.
   2. Seleziona Cloud Pub/Sub.
   3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento Pub/Sub netflow-trigger.
   4. Fai clic su Salva.

6. Nella sezione Autenticazione:

   1. Seleziona Richiedi autenticazione.
   2. Seleziona Identity and Access Management (IAM).
   Nota: Pub/Sub gestirà automaticamente l'autenticazione quando richiama la funzione.

7. Scorri verso il basso ed espandi Container, networking, sicurezza.

8. Vai alla scheda Sicurezza:

   • Service account: seleziona il account di servizio netflow-collector-sa

9. Vai alla scheda Container:

   1. Fai clic su Variabili e secret.
   2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:

   Nome variabile	Valore di esempio	Descrizione
   GCS_BUCKET	netflow-logs-bucket	Nome bucket GCS
   GCS_PREFIX	netflow	Prefisso per i file di log
   NETFLOW_PORT	2055	Porta per il ricevitore NetFlow
   NETFLOW_SCHEME	netflow	Tipo di schema: netflow, sflow o ipfix
   NETFLOW_SOCKETS	4	Numero di socket UDP
   NETFLOW_WORKERS	8	Numero di worker del decodificatore

10. Nella sezione Variabili e secret, scorri verso il basso fino a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti)

11. Vai alla scheda Impostazioni:

    • Nella sezione Risorse:
      • Memoria: seleziona 1 GiB o un valore superiore
      • CPU: seleziona 2.

12. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 1
    • Numero massimo di istanze: inserisci 10

13. Fai clic su Crea.

14. Attendi la creazione del servizio (1-2 minuti).

15. Dopo aver creato il servizio, si aprirà automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

1. Inserisci main nel campo Entry point (Punto di ingresso).

2. Nell'editor di codice incorporato, crea tre file:

   • 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. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

4. Attendi il completamento del deployment (3-5 minuti).

Nota: la funzione eseguirà un raccoglitore OpenTelemetry che ascolta i dati NetFlow e li esporta in GCS.

Configurare i dispositivi di rete per inviare dati NetFlow

Configura i tuoi dispositivi di rete (router, switch, firewall) per esportare i dati netflow, sflow o IPFIX nella funzione Cloud Run.

Recupera l'IP esterno della funzione Cloud Run

1. Nella console Google Cloud, vai a Cloud Run > Servizi.
2. Fai clic sul nome della funzione netflow-collector.
3. Copia l'URL visualizzato in alto (ad esempio, https://netflow-collector-xxxxx-uc.a.run.app).
4. Estrai il nome host dall'URL.

5. Utilizza nslookup o dig per risolvere l'indirizzo IP:

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

Nota: i servizi Cloud Run utilizzano IP dinamici. Per i deployment di produzione, valuta la possibilità di utilizzare un bilanciatore del carico Cloud con un IP statico o di eseguire il deployment del raccoglitore su Compute Engine.

• Esempio: configurazione NetFlow del router 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
  

• Esempio: configurazione NetFlow generica

  Per la maggior parte dei dispositivi di rete, configura le seguenti impostazioni:

  • Versione NetFlow: NetFlow v5, v9 o IPFIX
  • IP collector: indirizzo IP della funzione Cloud Run
  • Porta dell'agente di raccolta: 2055 (o porta configurata)
  • Protocollo: UDP
  • Timeout attivo: 60 secondi (consigliato)
  • Timeout di inattività: 15 secondi (consigliato)

Crea job Cloud Scheduler

Cloud Scheduler pubblicherà messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

1. Nella console Google Cloud, vai a Cloud Scheduler.
2. Fai clic su Crea job.

3. Fornisci i seguenti dettagli di configurazione:

   Impostazione	Valore
   Nome	netflow-collector-hourly
   Regione	Seleziona la stessa regione della funzione Cloud Run
   Frequenza	0 * * * * (ogni ora, allo scoccare dell'ora)
   Fuso orario	Seleziona il fuso orario (UTC consigliato)
   Tipo di target	Pub/Sub
   Argomento	Seleziona l'argomento Pub/Sub netflow-trigger
   Corpo del messaggio	{} (oggetto JSON vuoto)

4. Fai clic su Crea.

Opzioni di frequenza di pianificazione

Scegli la frequenza in base al volume dei log e ai requisiti di latenza:

Frequenza	Espressione cron	Caso d'uso
Ogni 5 minuti	*/5 * * * *	Volume elevato, bassa latenza
Ogni 15 minuti	*/15 * * * *	Volume medio
Ogni ora	0 * * * *	Standard (consigliato)
Ogni 6 ore	0 */6 * * *	Volume basso, elaborazione batch

Testare l'integrazione

1. Nella console Cloud Scheduler, trova il job netflow-collector-hourly.
2. Fai clic su Forza esecuzione per attivare il job manualmente.
3. Attendi alcuni secondi.
4. Vai a Cloud Run > Servizi.
5. Fai clic sul nome della funzione netflow-collector.
6. Fai clic sulla scheda Log.

7. Verifica che la funzione sia stata eseguita correttamente. Cerca:

   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. Vai a Cloud Storage > Bucket.

9. Fai clic sul nome del bucket netflow-logs-bucket.

10. Vai alla cartella del prefisso netflow/.

11. Verifica che sia stato creato un nuovo file .ndjson con il timestamp corrente.

Se nei log vengono visualizzati errori:

• Collector exited unexpectedly: controlla la configurazione del ricevitore NetFlow
• Nessun dato da caricare: verifica che i dispositivi di rete inviino i dati NetFlow all'IP e alla porta corretti
• Autorizzazione negata: verifica che il account di servizio abbia il ruolo Amministratore oggetti Storage
• Errore di binding della porta: assicurati che la porta 2055 non sia già in uso

Recuperare il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal tuo bucket GCS. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'email del account di servizio

1. Vai a Impostazioni SIEM > Feed.
2. Fai clic su Aggiungi nuovo feed.
3. Fai clic su Configura un singolo feed.
4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio OpenTelemetry Netflow Logs).
5. Seleziona Google Cloud Storage V2 come Tipo di origine.
6. Seleziona NETFLOW_OTEL come Tipo di log.

7. Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email univoco del account di servizio. Ad esempio:

   [e-mail]
   

8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

   Nota :ogni istanza di Google SecOps ha un account di servizio univoco. Non utilizzare i service account di altra documentazione o di altri esempi.

9. Fai clic su Avanti.

10. Specifica i valori per i seguenti parametri di input:

    • URL bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:

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

    Nota: includi sempre la barra finale (/) alla fine dell'URI.

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
      • Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.

      • Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.

        Nota: se selezioni un'opzione di eliminazione, il account di servizio deve disporre del ruolo Storage Object Admin anziché Storage Object Viewer. Aggiorna le autorizzazioni IAM di conseguenza.

    • Età massima del file: includi i file modificati nell'ultimo numero di giorni (il valore predefinito è 180 giorni)

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed

11. Fai clic su Avanti.

12. Controlla la nuova configurazione del feed di notizie nella schermata Finalizza, quindi fai clic su Invia.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel tuo bucket GCS.

1. Vai a Cloud Storage > Bucket.
2. Fai clic sul nome del bucket netflow-logs-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
   • Assegna i ruoli: seleziona Storage Object Viewer
6. Fai clic su Salva.

Nota: se prevedi di utilizzare l'opzione di eliminazione "Elimina file trasferiti" o "Elimina file trasferiti e directory vuote", assegna il ruolo Storage Object Admin anziché Storage Object Viewer.

Tabella di mappatura UDM

La tabella seguente mostra in che modo i campi dei log del ricevitore Netflow di OpenTelemetry vengono mappati ai campi UDM di Google SecOps:

Campo OpenTelemetry	Campo UDM	Descrizione
source.address	principal.ip	Indirizzo IP di origine
source.port	principal.port	Numero porta di origine
destination.address	target.ip	Indirizzo IP di destinazione
destination.port	target.port	Numero di porta di destinazione
network.transport	network.ip_protocol	Protocollo di trasporto (tcp, udp)
network.type	network.ip_version	Versione IP (ipv4, ipv6)
flow.io.bytes	network.sent_bytes	Byte totali trasferiti
flow.io.packets	network.sent_packets	Pacchetti totali trasferiti
flow.type	metadata.product_log_id	Tipo di flusso (NETFLOW_V5, NETFLOW_V9, SFLOW_5, IPFIX)
flow.start	metadata.event_timestamp	Ora di inizio del flusso
flow.end	network.session_duration	Ora di fine del flusso
flow.sampler_address	intermediary.ip	Indirizzo IP dell'esportatore NetFlow
flow.tcp_flags	network.tcp_flags	Flag TCP
flow.io.bytes	event.idm.read_only_udm.network.sent_bytes	Mappato dal log delle modifiche
source.port	event.idm.read_only_udm.principal.port	Mappato dal log delle modifiche
destination.port	event.idm.read_only_udm.target.port	Mappato dal log delle modifiche
network.transport	event.idm.read_only_udm.network.ip_protocol	Mappato dal log delle modifiche
network.type	event.idm.read_only_udm.security_result.detection_fields	Mappato dal log delle modifiche
flow.sequence_num	event.idm.read_only_udm.network.session_id	Mappato dal log delle modifiche
flow.sampler_address	event.idm.read_only_udm.observer.ip	Mappato dal log delle modifiche
flow.type	event.idm.read_only_udm.metadata.product_event_type	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.