Cisco ACI-Logs (Application Centric Infrastructure) erfassen
Parserversion: 4.0
In diesem Dokument wird beschrieben, wie Sie Cisco Application Centric Infrastructure (ACI)-Logs in Google Security Operations aufnehmen. Der Parser versucht zuerst, eingehende Cisco ACI-Logs als Syslog-Nachrichten mithilfe von Grok-Mustern zu verarbeiten. Wenn das Syslog-Parsing fehlschlägt, wird davon ausgegangen, dass die Nachricht im JSON-Format vorliegt, und sie wird entsprechend geparst. Schließlich werden die extrahierten Felder dem Unified Data Model (UDM) zugeordnet.
Diese Integration unterstützt zwei Methoden:
- Option 1: Syslog-Format mit Bindplane-Agent
- Option 2: JSON-Format für Google Cloud Storage mit der APIC REST API
Jede Option ist in sich abgeschlossen und kann unabhängig voneinander implementiert werden, je nach Ihren Infrastrukturanforderungen und bevorzugten Logformaten.
Option 1: Syslog mit dem Bindplane-Agent
Mit dieser Option wird das Cisco ACI-Fabric so konfiguriert, dass Syslog-Nachrichten an einen Bindplane-Agent gesendet werden, der sie zur Analyse an Google Security Operations weiterleitet.
Hinweis
Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:
- Eine Google SecOps-Instanz
- Ein Windows 2016- oder höher- oder Linux-Host mit
systemd - Wenn Sie den Agent hinter einem Proxy ausführen, müssen die Firewallports gemäß den Anforderungen des Bindplane-Agents geöffnet sein.
- Privilegierter Zugriff auf die Cisco APIC-Konsole
Google SecOps-Anmeldedaten für den Bindplane-Agenten abrufen
Wählen Sie eine der folgenden Methoden aus, um den Bindplane-Agent bei Google SecOps zu authentifizieren:
- Neue Bereitstellungen: Verwenden Sie die Chronicle API (Option 1).
- Alte Bereitstellungen: Verwenden Sie die Datei für die Authentifizierung bei der Aufnahme (Option 2) weiterhin, bis die Ingestion API am 20. Juli 2027 eingestellt wird.
Option 1: Dienstkonto für die Chronicle API erstellen (empfohlen)
- Melden Sie sich in der Google Cloud Console an und wählen Sie das Google Cloud -Projekt aus, das mit Ihrer Google SecOps-Instanz verknüpft ist.
- Rufen Sie IAM und Verwaltung > Dienstkonten auf.
- Klicken Sie auf Dienstkonto erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name des Dienstkontos: Geben Sie einen Namen für das Konto ein, z. B.
bindplane-ingest. - Dienstkontobeschreibung: Geben Sie eine Beschreibung ein (z. B.
Bindplane agent log ingestion).
- Name des Dienstkontos: Geben Sie einen Namen für das Konto ein, z. B.
- Klicken Sie auf Erstellen und fortfahren.
- Klicken Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen auf Rolle auswählen und wählen Sie Chronicle-Bearbeiter aus.
- Klicken Sie auf Weiter und dann auf Fertig.
- Klicken Sie auf das erstellte Dienstkonto und rufen Sie dann den Tab Schlüssel auf.
- Klicken Sie auf Schlüssel hinzufügen > Neuen Schlüssel erstellen.
- Wählen Sie JSON aus und klicken Sie auf Erstellen. Die Schlüsseldatei wird auf Ihren Computer heruntergeladen.
- Speichern Sie die Schlüsseldatei als
ingestion-auth.jsonim Bindplane-Collector-Verzeichnis (/opt/observiq-otel-collector/ingestion-auth.jsonunter Linux oderC:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.jsonunter Windows) oder aktualisieren Siecreds_file_pathinconfig.yaml, sodass der Pfad mit dem Pfad übereinstimmt, unter dem Sie die Datei speichern. - Rufen Sie IAM & Verwaltung > Einstellungen auf und kopieren Sie die Projektnummer. Sie benötigen sie für die Exporter-Konfiguration.
Option 2: Authentifizierungsdatei für die Datenaufnahme herunterladen (Legacy)
- Melden Sie sich in der Google SecOps-Konsole an.
- Rufen Sie die SIEM-Einstellungen > Erfassungs-Agents auf.
- Laden Sie die Authentifizierungsdatei für die Datenaufnahme herunter.
- Speichern Sie die Datei sicher auf dem System, auf dem der Bindplane-Agent installiert werden soll.
Google SecOps-Kundennummer abrufen
- Rufen Sie die SIEM-Einstellungen > Profile auf.
- Kopieren und speichern Sie die Kundennummer aus dem Bereich Organization Details (Organisationsdetails).
Bindplane-Agent installieren
Installieren Sie den Bindplane-Agent auf Ihrem Windows- oder Linux-Betriebssystem gemäß der folgenden Anleitung.
Windows-Installation
- Öffnen Sie die Eingabeaufforderung oder PowerShell als Administrator.
Führen Sie dazu diesen Befehl aus:
msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quietWarten Sie, bis die Installation abgeschlossen ist.
Überprüfen Sie die Installation mit folgendem Befehl:
sc query observiq-otel-collectorDer Dienst sollte als RUNNING angezeigt werden.
Linux-Installation
- Öffnen Sie ein Terminal mit Root- oder sudo-Berechtigungen.
Führen Sie dazu diesen Befehl aus:
sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.shWarten Sie, bis die Installation abgeschlossen ist.
Überprüfen Sie die Installation mit folgendem Befehl:
sudo systemctl status observiq-otel-collectorDer Dienst sollte als active (running) angezeigt werden.
Zusätzliche Installationsressourcen
Weitere Installationsoptionen und Informationen zur Fehlerbehebung finden Sie im Installationsanleitung für den Bindplane-Agenten.
Bindplane-Agent so konfigurieren, dass Syslog-Daten aufgenommen und an Google SecOps gesendet werden
Auf die Konfigurationsdatei zugreifen
- Suchen Sie die Datei
config.yaml. In der Regel befindet sie sich unter Linux im Verzeichnis/opt/observiq-otel-collector/oder unter Windows im Installationsverzeichnis. - Öffnen Sie die Datei mit einem Texteditor (z. B.
nano,vioder Notepad).
Datei „config.yaml“ bearbeiten
Datei
config.yamlreceivers: udplog: # Replace the port and IP address as required listen_address: "0.0.0.0:514" exporters: chronicle/chronicle_w_labels: protocol: https # Region of your instance, for example us, europe, asia-southeast1 location: <LOCATION> # Project number of the Google Cloud project linked to the instance project: '<PROJECT_NUMBER>' compression: gzip # Adjust the path to the credentials file you downloaded creds_file_path: '/path/to/ingestion-authentication-file.json' # Replace with your actual customer ID customer_id: <CUSTOMER_ID> endpoint: chronicle.googleapis.com # Add optional ingestion labels for better organization log_type: 'CISCO_ACI' raw_log_field: body ingestion_labels: service: pipelines: logs/source0__chronicle_w_labels-0: receivers: - udplog exporters: - chronicle/chronicle_w_labels
Konfigurationsparameter
Ersetzen Sie die folgenden Platzhalter:
listen_address: Ersetzen Sie den Port und die IP-Adresse nach Bedarf in Ihrer Infrastruktur. Verwenden Sie0.0.0.0:514, um alle Schnittstellen an Port 514 abzuhören.creds_file_path: Aktualisieren Sie den Pfad, unter dem die Dienstkontoschlüsseldatei (Option 1) gespeichert wurde:- Linux:
/opt/observiq-otel-collector/ingestion-auth.json - Windows:
C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
- Linux:
customer_id: Ersetzen SieYOUR_CUSTOMER_IDdurch die tatsächliche Kundennummer aus dem vorherigen Schritt.location: Ersetzen Sie<LOCATION>durch die Region Ihrer Google SecOps-Instanz (z. B.us,europeoderasia-southeast1). Der Exporter stellt diesem Wertchronicle.googleapis.comvoran. Eine vollständige Liste finden Sie unter Regionale Endpunkte.project: Ersetzen Sie<PROJECT_NUMBER>durch die Projektnummer aus IAM & Verwaltung > Einstellungen.endpoint: Behalten Sie die Einstellungchronicle.googleapis.combei.
Legacy-Ingestion-API-Exporter
Wenn Sie die Authentifizierungsdatei für die Datenaufnahme aus Option 2 verwenden, ersetzen Sie den Block
exportersdurch Folgendes. Legen Siecreds_file_pathauf den Pfad zur Datei für die Authentifizierung bei der Aufnahme fest. Wenn sich Ihre Instanz außerhalb der USA befindet, ersetzen Siemalachiteingestion-pa.googleapis.comdurch Ihren regionalen Endpunkt (z. B.europe-malachiteingestion-pa.googleapis.com).exporters: chronicle/chronicle_w_labels: compression: gzip # Adjust the path to the credentials file you downloaded creds_file_path: '/path/to/ingestion-authentication-file.json' # Replace with your actual customer ID customer_id: <CUSTOMER_ID> endpoint: malachiteingestion-pa.googleapis.com # Add optional ingestion labels for better organization log_type: 'CISCO_ACI' raw_log_field: body ingestion_labels: service:
Konfigurationsdatei speichern
Speichern Sie die Datei nach der Bearbeitung:
- Linux: Drücken Sie
Ctrl+O, dannEnterund dannCtrl+X. - Windows: Klicken Sie auf Datei > Speichern.
Bindplane-Agent neu starten, um die Änderungen zu übernehmen
Linux
So starten Sie den Bindplane-Agent auf einem Linux-System neu:
Starten Sie den Agent neu:
sudo systemctl restart observiq-otel-collectorPrüfen Sie den Dienststatus:
sudo systemctl status observiq-otel-collectorAgent-Logs auf Fehler prüfen:
sudo journalctl -u observiq-otel-collector -f
Windows
Wählen Sie eine der folgenden Optionen aus, um den Bindplane-Agenten unter Windows neu zu starten:
Eingabeaufforderung oder PowerShell als Administrator:
net stop observiq-otel-collector && net start observiq-otel-collectorDienstkonsole:
- Drücken Sie
Win+R, geben Sieservices.mscein und drücken Sie die Eingabetaste. - Suchen Sie nach observIQ OpenTelemetry Collector.
- Klicken Sie mit der rechten Maustaste und wählen Sie Neu starten aus.
Prüfen Sie, ob der Dienst ausgeführt wird:
sc query observiq-otel-collectorLogs auf Fehler prüfen:
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
- Drücken Sie
Syslog-Weiterleitung in Cisco ACI konfigurieren
Out-of-Band-Managementvertrag konfigurieren
- Melden Sie sich in der Cisco APIC-Konsole an.
- Rufen Sie Tenants > mgmt> Contracts> Filters auf.
- Klicken Sie auf Filter erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
syslog-udp-514ein. - Name des Eintrags: Geben Sie
syslogein. - EtherType: Wählen Sie IP aus.
- IP-Protokoll: Wählen Sie UDP aus.
- Zielportbereich (von): Geben Sie
514ein. - Zielportbereich bis: Geben Sie
514ein.
- Name: Geben Sie
- Klicken Sie auf Senden.
Verwaltungsvertrag erstellen
- Gehen Sie zu Tenants > mgmt > Contracts > Standard.
- Klicken Sie auf Vertrag erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
mgmt-syslog-contractein. - Umfang: Wählen Sie Kontext aus.
- Name: Geben Sie
- Klicken Sie auf Senden.
- Maximieren Sie den Vertrag und klicken Sie auf Themen.
- Klicken Sie auf Vertragsgegenstand erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
syslog-subjectein. - Apply Both Directions (Beide Richtungen anwenden): Aktivieren Sie diese Option.
- Name: Geben Sie
- Klicken Sie auf Senden.
- Maximieren Sie das Thema und klicken Sie auf Filter.
- Klicken Sie auf Filterbindung erstellen.
- Wählen Sie den Filter
syslog-udp-514aus. - Klicken Sie auf Senden.
Syslog-Zielgruppe konfigurieren
- Gehen Sie zu Verwaltung > Externe Datenerfasser > Monitoring-Ziele > Syslog.
- Klicken Sie mit der rechten Maustaste auf Syslog und wählen Sie Create Syslog Monitoring Destination Group (Syslog-Überwachungszielgruppe erstellen) aus.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
Chronicle-Syslog-Groupein. - Admin-Status: Wählen Sie Aktiviert aus.
- Format: Wählen Sie aci aus.
- Name: Geben Sie
- Klicken Sie auf Weiter.
- Gehen Sie im Dialogfeld Syslog-Monitoringziel erstellen so vor:
- Name: Geben Sie
Chronicle-BindPlaneein. - Host: Geben Sie die IP-Adresse Ihres Bindplane-Agent-Servers ein.
- Port: Geben Sie
514ein. - Admin-Status: Wählen Sie Aktiviert aus.
- Schweregrad: Wählen Sie Informationen aus, um detaillierte Logs zu erfassen.
- Name: Geben Sie
- Klicken Sie auf Senden.
Monitoring-Richtlinien konfigurieren – Fabric Monitoring
- Rufen Sie Fabric > Fabric Policies > Policies > Monitoring > Common Policy auf.
- Maximieren Sie Callhome/Smart Callhome/SNMP/Syslog/TACACS.
- Klicken Sie mit der rechten Maustaste auf Syslog und wählen Sie Syslog-Quelle erstellen aus.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
Chronicle-Fabric-Syslogein. - Audit-Logs: Aktivieren Sie diese Option, um Audit-Ereignisse einzubeziehen.
- Ereignisse: Aktivieren Sie diese Option, um Systemereignisse einzubeziehen.
- Fehler: Aktivieren Sie diese Option, um Fehlerereignisse einzubeziehen.
- Sitzungsprotokolle: Setzen Sie ein Häkchen, um Sitzungsprotokolle einzuschließen.
- Zielgruppe: Wählen Sie
Chronicle-Syslog-Groupaus.
- Name: Geben Sie
- Klicken Sie auf Senden.
Monitoring-Richtlinien konfigurieren – Zugriff überwachen
- Gehen Sie zu Fabric > Access Policies > Policies > Monitoring > Default Policy.
- Maximieren Sie Callhome/Smart Callhome/SNMP/Syslog.
- Rechtsklicken Sie auf Syslog und wählen Sie Syslog-Quelle erstellen aus.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
Chronicle-Access-Syslogein. - Audit-Logs: Aktivieren Sie diese Option, um Audit-Ereignisse einzubeziehen.
- Ereignisse: Aktivieren Sie diese Option, um Systemereignisse einzubeziehen.
- Fehler: Aktivieren Sie diese Option, um Fehlerereignisse einzubeziehen.
- Sitzungsprotokolle: Setzen Sie ein Häkchen, um Sitzungsprotokolle einzuschließen.
- Zielgruppe: Wählen Sie
Chronicle-Syslog-Groupaus.
- Name: Geben Sie
- Klicken Sie auf Senden.
Richtlinie für System-Syslog-Nachrichten konfigurieren
- Gehen Sie zu Fabric > Fabric Policies > Policies > Monitoring > Common Policy.
- Maximieren Sie Syslog Messages Policies.
- Klicken Sie auf Standard.
- Im Abschnitt Einrichtungsfilter:
- Einrichtung: Wählen Sie Standard aus.
- Mindestschweregrad: Ändern Sie den Wert in Information.
- Klicken Sie auf Senden.
Option 2: JSON-Dateien über Google Cloud Storage
Bei dieser Option werden über die APIC REST API JSON-formatierte Ereignisse, Fehler und Audit-Logs aus der Cisco ACI-Fabric erfasst und in Google Cloud Storage für die Aufnahme in Google SecOps gespeichert.
Hinweis
Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:
- Eine Google SecOps-Instanz
- Ein Google Cloud -Projekt mit aktivierter Cloud Storage API
- Berechtigungen zum Erstellen und Verwalten von GCS-Buckets
- Berechtigungen zum Verwalten von IAM-Richtlinien für GCS-Buckets
- Berechtigungen zum Erstellen von Cloud Run-Diensten, Pub/Sub-Themen und Cloud Scheduler-Jobs
- Privilegierter Zugriff auf die Cisco APIC-Konsole
Voraussetzungen für Cisco ACI APIC erfassen
APIC-Anmeldedaten abrufen
- Melden Sie sich über HTTPS in der Cisco APIC-Konsole an.
Gehen Sie zu Admin > AAA (auf APIC 6.0 oder höher) oder Admin > Authentication > AAA (auf älteren Versionen).
Erstellen Sie einen neuen lokalen Nutzer mit den entsprechenden Berechtigungen oder verwenden Sie einen vorhandenen.
Kopieren und speichern Sie die folgenden Details an einem sicheren Ort:
- APIC-Nutzername: Lokaler Nutzer mit Lesezugriff auf Monitoring-Daten
- APIC-Passwort: Nutzerpasswort
- APIC-URL: Die HTTPS-URL Ihres APIC (z. B.
https://apic.example.com)
Google Cloud Storage-Bucket erstellen
- Rufen Sie die Google Cloud Console auf.
- Wählen Sie Ihr Projekt aus oder erstellen Sie ein neues.
- Rufen Sie im Navigationsmenü Cloud Storage > Buckets auf.
- Klicken Sie auf Bucket erstellen.
Geben Sie die folgenden Konfigurationsdetails an:
Einstellung Wert Bucket benennen Geben Sie einen global eindeutigen Namen ein, z. B. cisco-aci-logs.Standorttyp Wählen Sie die Option, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region). Standort Wählen Sie den Standort aus (z. B. us-central1).Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird) Zugriffssteuerung Einheitlich (empfohlen) Schutztools Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren Klicken Sie auf Erstellen.
Dienstkonto für Cloud Run-Funktion erstellen
Für die Cloud Run-Funktion ist ein Dienstkonto mit Berechtigungen zum Schreiben in den GCS-Bucket und zum Aufrufen durch Pub/Sub erforderlich.
Dienstkonto erstellen
- Wechseln Sie in der GCP Console zu IAM und Verwaltung > Dienstkonten.
- Klicken Sie auf Dienstkonto erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Dienstkontoname: Geben Sie
cisco-aci-collector-saein. - Beschreibung des Dienstkontos: Geben Sie
Service account for Cloud Run function to collect Cisco ACI logsein.
- Dienstkontoname: Geben Sie
- Klicken Sie auf Erstellen und fortfahren.
- Fügen Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen die folgenden Rollen hinzu:
- Klicken Sie auf Rolle auswählen.
- Suchen Sie nach Storage Object Admin und wählen Sie die Rolle aus.
- Klicken Sie auf + Weitere Rolle hinzufügen.
- Suchen Sie nach Cloud Run Invoker und wählen Sie die Rolle aus.
- Klicken Sie auf + Weitere Rolle hinzufügen.
- Suchen Sie nach Cloud Functions-Aufrufer und wählen Sie diese Option aus.
- Klicken Sie auf Weiter.
- Klicken Sie auf Fertig.
Diese Rollen sind für Folgendes erforderlich:
- Storage-Objekt-Administrator: Logs in den GCS-Bucket schreiben und Zustandsdateien verwalten
- Cloud Run-Aufrufer: Pub/Sub erlauben, die Funktion aufzurufen
- Cloud Functions-Invoker: Funktionsaufruf zulassen
IAM-Berechtigungen für GCS-Bucket erteilen
Gewähren Sie dem Dienstkonto (cisco-aci-collector-sa) Schreibberechtigungen für den GCS-Bucket:
- Rufen Sie Cloud Storage > Buckets auf.
- Klicken Sie auf den Namen Ihres Buckets (z. B.
cisco-aci-logs). - Wechseln Sie zum Tab Berechtigungen.
- Klicken Sie auf Zugriff erlauben.
- Geben Sie die folgenden Konfigurationsdetails an:
- Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos ein (z. B.
cisco-aci-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Rollen zuweisen: Wählen Sie Storage Object Admin aus.
- Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos ein (z. B.
- Klicken Sie auf Speichern.
Pub/Sub-Thema erstellen
Erstellen Sie ein Pub/Sub-Thema, in dem Cloud Scheduler veröffentlicht und das von der Cloud Run-Funktion abonniert wird.
- Wechseln Sie in der GCP Console zu Pub/Sub > Themen.
- Klicken Sie auf Thema erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Themen-ID: Geben Sie
cisco-aci-triggerein. - Übernehmen Sie die anderen Einstellungen.
- Themen-ID: Geben Sie
- Klicken Sie auf Erstellen.
Cloud Run-Funktion zum Erfassen von Logs erstellen
Die Cloud Run-Funktion wird durch Pub/Sub-Nachrichten von Cloud Scheduler ausgelöst, um Logs von der Cisco APIC REST API abzurufen und in GCS zu schreiben.
- Rufen Sie in der GCP Console die Option Cloud Run auf.
- Klicken Sie auf Dienst erstellen.
- Wählen Sie Funktion aus (Funktion mit einem Inline-Editor erstellen).
Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:
Einstellung Wert Dienstname cisco-aci-collectorRegion Wählen Sie die Region aus, die Ihrem GCS-Bucket entspricht (z. B. us-central1).Laufzeit Wählen Sie Python 3.12 oder höher aus. Im Abschnitt Trigger (optional):
- Klicken Sie auf + Trigger hinzufügen.
- Wählen Sie Cloud Pub/Sub aus.
- Wählen Sie unter Cloud Pub/Sub-Thema auswählen das Pub/Sub-Thema (
cisco-aci-trigger) aus. - Klicken Sie auf Speichern.
Im Abschnitt Authentifizierung:
- Wählen Sie Authentifizierung erforderlich aus.
- Prüfen Sie Identity and Access Management (IAM).
Scrollen Sie nach unten und maximieren Sie Container, Netzwerk, Sicherheit.
Rufen Sie den Tab Sicherheit auf:
- Dienstkonto: Wählen Sie das Dienstkonto aus (
cisco-aci-collector-sa).
- Dienstkonto: Wählen Sie das Dienstkonto aus (
Rufen Sie den Tab Container auf:
- Klicken Sie auf Variablen und Secrets.
- Klicken Sie für jede Umgebungsvariable auf + Variable hinzufügen:
Variablenname Beispielwert Beschreibung GCS_BUCKETcisco-aci-logsName des GCS-Buckets GCS_PREFIXcisco-aci-eventsPräfix für Logdateien STATE_KEYcisco-aci-events/state.jsonPfad der Statusdatei APIC_URLhttps://apic.example.comAPIC-HTTPS-URL APIC_USERNAMEyour-apic-usernameAPIC-Nutzername APIC_PASSWORDyour-apic-passwordAPIC-Passwort PAGE_SIZE100Einträge pro Seite MAX_PAGES10Maximale Anzahl von Seiten pro Ausführung Scrollen Sie im Bereich Variablen und Secrets zu Anfragen:
- Zeitüberschreitung bei Anfrage: Geben Sie
600Sekunden (10 Minuten) ein.
- Zeitüberschreitung bei Anfrage: Geben Sie
Rufen Sie den Tab Einstellungen auf:
- Im Abschnitt Ressourcen:
- Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
- CPU: Wählen Sie 1 aus.
- Im Abschnitt Ressourcen:
Im Abschnitt Versionsskalierung:
- Mindestanzahl von Instanzen: Geben Sie
0ein. - Maximale Anzahl von Instanzen: Geben Sie
100ein oder passen Sie den Wert an die erwartete Last an.
- Mindestanzahl von Instanzen: Geben Sie
Klicken Sie auf Erstellen.
Warten Sie ein bis zwei Minuten, bis der Dienst erstellt wurde.
Nachdem der Dienst erstellt wurde, wird der Inline-Code-Editor automatisch geöffnet.
Funktionscode hinzufügen
- Geben Sie main in das Feld Einstiegspunkt ein.
Erstellen Sie im Inline-Code-Editor zwei Dateien:
- Erste Datei: main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import logging # Configure logging logger = logging.getLogger() logger.setLevel(logging.INFO) # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=60.0), retries=False, ) # Initialize Storage client storage_client = storage.Client() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'cisco-aci-events') STATE_KEY = os.environ.get('STATE_KEY', 'cisco-aci-events/state.json') APIC_URL = os.environ.get('APIC_URL') APIC_USERNAME = os.environ.get('APIC_USERNAME') APIC_PASSWORD = os.environ.get('APIC_PASSWORD') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '10')) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Cisco ACI logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, APIC_URL, APIC_USERNAME, APIC_PASSWORD]): logger.error('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window last_timestamp = state.get('last_timestamp') if not last_timestamp: last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z' logger.info(f"Starting Cisco ACI data collection for bucket: {GCS_BUCKET}") # Authenticate to APIC session_token = authenticate_apic(APIC_URL, APIC_USERNAME, APIC_PASSWORD) headers = { 'Cookie': f'APIC-cookie={session_token}', 'Accept': 'application/json', 'Content-Type': 'application/json' } # Data types to collect data_types = ['faultInst', 'eventRecord', 'aaaModLR'] all_collected_data = [] for data_type in data_types: logger.info(f"Collecting {data_type} data") collected_data = collect_aci_data( APIC_URL, headers, data_type, last_timestamp, PAGE_SIZE, MAX_PAGES ) # Tag each record with its type for record in collected_data: record['_data_type'] = data_type all_collected_data.extend(collected_data) logger.info(f"Collected {len(collected_data)} {data_type} records") logger.info(f"Total records collected: {len(all_collected_data)}") # Store data in GCS if any were collected if all_collected_data: timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S') s3_key = f"{GCS_PREFIX}/cisco_aci_events_{timestamp_str}.ndjson" # Convert to NDJSON format (one JSON object per line) ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data) # Upload to GCS blob = bucket.blob(s3_key) blob.upload_from_string( ndjson_content, content_type='application/x-ndjson' ) logger.info(f"Uploaded {len(all_collected_data)} records to gs://{GCS_BUCKET}/{s3_key}") # Update state file with latest timestamp from collected data latest_timestamp = get_latest_timestamp_from_records(all_collected_data) if not latest_timestamp: latest_timestamp = datetime.utcnow().isoformat() + 'Z' update_state(bucket, STATE_KEY, latest_timestamp) else: logger.info("No new log records found.") logger.info(f"Successfully processed {len(all_collected_data)} records") except Exception as e: logger.error(f'Error processing logs: {str(e)}') raise def authenticate_apic(apic_url, username, password): """Authenticate to APIC and return session token""" login_url = f"{apic_url}/api/aaaLogin.json" login_data = { "aaaUser": { "attributes": { "name": username, "pwd": password } } } response = http.request( 'POST', login_url, body=json.dumps(login_data).encode('utf-8'), headers={'Content-Type': 'application/json'}, timeout=30 ) if response.status != 200: raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}") response_data = json.loads(response.data.decode('utf-8')) token = response_data['imdata'][0]['aaaLogin']['attributes']['token'] logger.info("Successfully authenticated to APIC") return token def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages): """Collect data from APIC REST API with pagination""" all_data = [] page = 0 while page < max_pages: # Build API URL with pagination and time filters api_url = f"{apic_url}/api/class/{data_type}.json" params = [ f'page-size={page_size}', f'page={page}', f'order-by={data_type}.created|asc' ] # Add time filter to prevent duplicates if last_timestamp: params.append(f'query-target-filter=gt({data_type}.created,"{last_timestamp}")') full_url = f"{api_url}?{'&'.join(params)}" logger.info(f"Fetching {data_type} page {page} from APIC") # Make API request response = http.request('GET', full_url, headers=headers, timeout=60) if response.status != 200: logger.error(f"API request failed: {response.status} {response.data[:256]!r}") break data = json.loads(response.data.decode('utf-8')) records = data.get('imdata', []) if not records: logger.info(f"No more {data_type} records found") break # Extract the actual data from APIC format extracted_records = [] for record in records: if data_type in record: extracted_records.append(record[data_type]) all_data.extend(extracted_records) page += 1 # If we got less than page_size records, we've reached the end if len(records) < page_size: break return all_data def get_last_timestamp(bucket, state_key): """Get the last run timestamp from GCS state file""" try: blob = bucket.blob(state_key) if blob.exists(): state_data = blob.download_as_text() state = json.loads(state_data) return state.get('last_timestamp') except Exception as e: logger.warning(f"Error reading state file: {str(e)}") return None def get_latest_timestamp_from_records(records): """Get the latest timestamp from collected records to prevent missing events""" if not records: return None latest = None latest_time = None for record in records: try: # Handle both direct attributes and nested structure attrs = record.get('attributes', record) created = attrs.get('created') modTs = attrs.get('modTs') # Use created or modTs as fallback timestamp = created or modTs if timestamp: if latest_time is None or timestamp > latest_time: latest_time = timestamp latest = record except Exception as e: logger.debug(f"Error parsing timestamp from record: {e}") continue return latest_time def update_state(bucket, state_key, timestamp): """Update the state file with the current timestamp""" try: state_data = { 'last_timestamp': timestamp, 'updated_at': datetime.utcnow().isoformat() + 'Z' } blob = bucket.blob(state_key) blob.upload_from_string( json.dumps(state_data), content_type='application/json' ) logger.info(f"Updated state file with timestamp: {timestamp}") except Exception as e: logger.error(f"Error updating state file: {str(e)}") def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: logger.warning(f"Could not load state: {e}") return {}- Zweite Datei: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Klicken Sie auf Bereitstellen, um die Funktion zu speichern und bereitzustellen.
Warten Sie, bis die Bereitstellung abgeschlossen ist (2–3 Minuten).
Cloud Scheduler-Job erstellen
Cloud Scheduler veröffentlicht in regelmäßigen Abständen Nachrichten im Pub/Sub-Thema und löst so die Cloud Run-Funktion aus.
- Rufen Sie in der GCP Console Cloud Scheduler auf.
- Klicken Sie auf Job erstellen.
Geben Sie die folgenden Konfigurationsdetails an:
Einstellung Wert Name cisco-aci-collector-15mRegion Wählen Sie dieselbe Region wie für die Cloud Run-Funktion aus. Frequenz */15 * * * *(alle 15 Minuten)Zeitzone Zeitzone auswählen (UTC empfohlen) Zieltyp Pub/Sub Thema Pub/Sub-Thema auswählen ( cisco-aci-trigger)Inhalt der Nachricht {}(leeres JSON-Objekt)Klicken Sie auf Erstellen.
Optionen für die Häufigkeit des Zeitplans
Wählen Sie die Häufigkeit basierend auf dem Logvolumen und den Latenzanforderungen aus:
| Häufigkeit | Cron-Ausdruck | Anwendungsfall |
|---|---|---|
| Alle 5 Minuten | */5 * * * * |
Hohes Volumen, niedrige Latenz |
| Alle 15 Minuten | */15 * * * * |
Mittleres Suchvolumen (empfohlen) |
| Stündlich | 0 * * * * |
Standard |
| Alle 6 Stunden | 0 */6 * * * |
Geringes Volumen, Batchverarbeitung |
| Täglich | 0 0 * * * |
Erhebung von Verlaufsdaten |
Integration testen
- Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job (
cisco-aci-collector-15m). - Klicken Sie auf Ausführung erzwingen, um den Job manuell auszulösen.
- Warte einige Sekunden.
- Rufen Sie Cloud Run > Dienste auf.
- Klicken Sie auf den folgenden Funktionsnamen (
cisco-aci-collector). - Klicken Sie auf den Tab Logs.
Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde. Darauf sollten Sie achten:
Starting Cisco ACI data collection for bucket: cisco-aci-logs Successfully authenticated to APIC Collecting faultInst data Collected X faultInst records Collecting eventRecord data Collected X eventRecord records Collecting aaaModLR data Collected X aaaModLR records Total records collected: X Uploaded X records to gs://cisco-aci-logs/cisco-aci-events/cisco_aci_events_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsRufen Sie Cloud Storage > Buckets auf.
Klicken Sie auf den Namen Ihres Buckets (
cisco-aci-logs).Rufen Sie den Präfixordner (
cisco-aci-events/) auf.Prüfen Sie, ob eine neue
.ndjson-Datei mit dem aktuellen Zeitstempel erstellt wurde.
Wenn Sie Fehler in den Logs sehen:
- HTTP 401: APIC-Anmeldedaten in Umgebungsvariablen prüfen
- HTTP 403: Prüfen Sie, ob das APIC-Konto Leseberechtigungen für die Klassen
faultInst,eventRecordundaaaModLRhat. - Verbindungsfehler: Prüfen Sie, ob die Cloud Run-Funktion die APIC-URL über TCP/443 erreichen kann.
- Fehlende Umgebungsvariablen: Prüfen Sie, ob alle erforderlichen Variablen festgelegt sind.
Feed in Google SecOps konfigurieren, um Cisco ACI-Logs aufzunehmen
- Rufen Sie die SIEM-Einstellungen > Feeds auf.
- Klicken Sie auf Neuen Feed hinzufügen.
- Klicken Sie auf Einzelnen Feed konfigurieren.
- Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B.
Cisco ACI JSON logs. - Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
- Wählen Sie Cisco Application Centric Infrastructure als Log type (Logtyp) aus.
Klicken Sie auf Dienstkonto abrufen. Es wird eine eindeutige E-Mail-Adresse für das Dienstkonto angezeigt, z. B.:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comKopieren Sie diese E-Mail-Adresse. Sie verwenden es in der nächsten Aufgabe.
Klicken Sie auf Weiter.
Geben Sie Werte für die folgenden Eingabeparameter an:
- Storage-Bucket-URL: Geben Sie den GCS-Bucket-URI mit dem Präfixpfad ein:
gs://cisco-aci-logs/cisco-aci-events/- Ersetzen:
cisco-aci-logs: Der Name Ihres GCS-Buckets.cisco-aci-events: Präfix/Ordnerpfad, in dem Logs gespeichert werden.
- Option zum Löschen der Quelle: Wählen Sie die gewünschte Option zum Löschen aus:
- Nie: Nach Übertragungen werden keine Dateien gelöscht (für Tests empfohlen).
- Übertragene Dateien löschen: Dateien werden nach der erfolgreichen Übertragung gelöscht.
- Übertragene Dateien und leere Verzeichnisse löschen: Löscht Dateien und leere Verzeichnisse nach der erfolgreichen Übertragung.
- Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.
- Asset-Namespace: Der Asset-Namespace.
- Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet wird.
Klicken Sie auf Weiter.
Prüfen Sie die neue Nachrichten-Feed-Konfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.
Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren
Das Google SecOps-Dienstkonto benötigt die Rolle Storage Object Viewer für Ihren GCS-Bucket.
- Rufen Sie Cloud Storage > Buckets auf.
- Klicken Sie auf den Namen Ihres Buckets (
cisco-aci-logs). - Wechseln Sie zum Tab Berechtigungen.
- Klicken Sie auf Zugriff erlauben.
- Geben Sie die folgenden Konfigurationsdetails an:
- Hauptkonten hinzufügen: Fügen Sie die E‑Mail-Adresse des Google SecOps-Dienstkontos ein.
- Rollen zuweisen: Wählen Sie Storage Object Viewer aus.
Klicken Sie auf Speichern.
UDM-Zuordnungstabelle
| Logfeld | UDM-Zuordnung | Logik |
|---|---|---|
| @timestamp | read_only_udm.metadata.event_timestamp | Der Wert wird aus dem Rohlogfeld „@timestamp“ übernommen und als Zeitstempel geparst. |
| aci_tag | read_only_udm.metadata.product_log_id | Der Wert stammt aus dem Rohlogfeld „aci_tag“. |
| cisco_timestamp | - | Nicht zugeordnet. |
| DIP | read_only_udm.target.ip | Der Wert wird aus dem Rohlogfeld „DIP“ übernommen. |
| DPort | read_only_udm.target.port | Der Wert wird aus dem Rohlogfeld „DPort“ übernommen und in eine Ganzzahl konvertiert. |
| Beschreibung | read_only_udm.security_result.description | Der Wert wird aus dem Rohlogfeld „description“ übernommen. |
| fault_cause | read_only_udm.additional.fields.value.string_value | Der Wert wird aus dem Rohlogfeld „fault_cause“ übernommen. Der Schlüssel ist auf „Fault Cause“ (Fehlerursache) festgelegt. |
| Hostname | read_only_udm.principal.hostname | Der Wert wird aus dem Rohlogfeld „hostname“ übernommen. |
| lifecycle_state | read_only_udm.metadata.product_event_type | Der Wert stammt aus dem Rohlogfeld „lifecycle_state“. |
| log.source.address | - | Nicht zugeordnet. |
| logstash.collect.host | - | Nicht zugeordnet. |
| logstash.collect.timestamp | read_only_udm.metadata.collected_timestamp | Der Wert wird aus dem Rohlogfeld „logstash.collect.timestamp“ übernommen und als Zeitstempel geparst. |
| logstash.ingest.host | read_only_udm.intermediary.hostname | Der Wert wird aus dem Rohlogfeld „logstash.ingest.host“ übernommen. |
| logstash.irm_environment | read_only_udm.additional.fields.value.string_value | Der Wert wird aus dem Rohlogfeld „logstash.irm_environment“ übernommen. Der Schlüssel wird auf „IRM_Environment“ festgelegt. |
| logstash.irm_region | read_only_udm.additional.fields.value.string_value | Der Wert wird aus dem Rohlogfeld „logstash.irm_region“ übernommen. Der Schlüssel ist auf „IRM_Region“ festgelegt. |
| logstash.irm_site | read_only_udm.additional.fields.value.string_value | Der Wert wird aus dem Rohlogfeld „logstash.irm_site“ übernommen. Der Schlüssel ist auf „IRM_Site“ festgelegt. |
| logstash.process.host | read_only_udm.intermediary.hostname | Der Wert wird aus dem Rohlogfeld „logstash.process.host“ übernommen. |
| Nachricht | - | Nicht zugeordnet. |
| message_class | - | Nicht zugeordnet. |
| message_code | - | Nicht zugeordnet. |
| message_content | - | Nicht zugeordnet. |
| message_dn | - | Nicht zugeordnet. |
| message_type | read_only_udm.metadata.product_event_type | Der Wert wird aus dem Rohlogfeld „message_type“ übernommen, nachdem die eckigen Klammern entfernt wurden. |
| node_link | read_only_udm.principal.process.file.full_path | Der Wert stammt aus dem Rohlogfeld „node_link“. |
| PktLen | read_only_udm.network.received_bytes | Der Wert wird aus dem Rohlogfeld „PktLen“ übernommen und in eine vorzeichenlose Ganzzahl konvertiert. |
| über das Programm | - | Nicht zugeordnet. |
| Proto | read_only_udm.network.ip_protocol | Der Wert wird aus dem Rohlogfeld „Proto“ übernommen, in eine Ganzzahl umgewandelt und dem entsprechenden IP-Protokollnamen zugeordnet (z.B. 6 –> TCP). |
| SIP | read_only_udm.principal.ip | Der Wert wird aus dem Rohlogfeld „SIP“ übernommen. |
| SPort | read_only_udm.principal.port | Der Wert wird aus dem Rohlogfeld „SPort“ übernommen und in eine Ganzzahl konvertiert. |
| syslog_facility | - | Nicht zugeordnet. |
| syslog_facility_code | - | Nicht zugeordnet. |
| syslog_host | read_only_udm.principal.ip, read_only_udm.observer.ip | Der Wert wird aus dem Rohlogfeld „syslog_host“ übernommen. |
| syslog_prog | - | Nicht zugeordnet. |
| syslog_severity | read_only_udm.security_result.severity_details | Der Wert wird aus dem Rohprotokollfeld „syslog_severity“ übernommen. |
| syslog_severity_code | read_only_udm.security_result.severity | Der Wert wird aus dem Rohlogfeld „syslog_severity_code“ übernommen und dem entsprechenden Schweregrad zugeordnet: 5, 6, 7 –> INFORMATIONAL; 3, 4 –> MEDIUM; 0, 1, 2 –> HIGH. |
| syslog5424_pri | - | Nicht zugeordnet. |
| VLAN-ID | read_only_udm.principal.resource.id | Der Wert wird aus dem Rohlogfeld „Vlan-Id“ übernommen. |
| - | read_only_udm.metadata.event_type | Logik: Wenn „SIP“ oder „hostname“ und „Proto“ vorhanden sind, wird der Wert auf „NETWORK_CONNECTION“ gesetzt. Andernfalls, wenn „SIP“, „hostname“ oder „syslog_host“ vorhanden ist, wird der Wert auf „STATUS_UPDATE“ gesetzt. Andernfalls legen Sie „GENERIC_EVENT“ fest. |
| - | read_only_udm.metadata.log_type | Logik: Auf „CISCO_ACI“ festgelegt. |
| - | read_only_udm.metadata.vendor_name | Logik: Auf „Cisco“ festlegen. |
| - | read_only_udm.metadata.product_name | Logik: Auf „ACI“ festgelegt. |
principal_mac |
event.idm.read_only_udm.principal.mac und event.idm.read_only_udm.principal.asset.mac |
Aus Änderungsprotokoll zugeordnet |
context_id |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
node_number |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
code1 |
event.idm.read_only_udm.security_result.detection_fields |
Aus Änderungsprotokoll zugeordnet |
code2 |
event.idm.read_only_udm.security_result.detection_fields |
Aus Änderungsprotokoll zugeordnet |
pid |
event.idm.read_only_udm.principal.process.pid |
Aus Änderungsprotokoll zugeordnet |
src_intf |
event.idm.read_only_udm.principal.resource.attribute.labels |
Aus Änderungsprotokoll zugeordnet |
VXLAN |
event.idm.read_only_udm.network.session_id |
Aus Änderungsprotokoll zugeordnet |
VlanType |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
flag |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
Hitcnt |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
syslog_facility |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
SMac |
event.idm.read_only_udm.principal.resource.attribute.labels |
Aus Änderungsprotokoll zugeordnet |
DMac |
event.idm.read_only_udm.target.resource.attribute.labels |
Aus Änderungsprotokoll zugeordnet |
path |
event.idm.read_only_udm.principal.process.file.full_path |
Aus Änderungsprotokoll zugeordnet |
CName |
event.idm.read_only_udm.security_result.rule_name |
Aus Änderungsprotokoll zugeordnet |
i_ip |
event.idm.read_only_udm.intermediary.ip |
Aus Änderungsprotokoll zugeordnet |
i_ip |
event.idm.read_only_udm.intermediary.asset.ip |
Aus Änderungsprotokoll zugeordnet |
log_id |
event.idm.read_only_udm.metadata.product_log_id |
Aus Änderungsprotokoll zugeordnet |
product_event_type |
event.idm.read_only_udm.metadata.product_event_type |
Aus Änderungsprotokoll zugeordnet |
severity_action |
event.idm.read_only_udm.security_result.severity |
Aus Änderungsprotokoll zugeordnet |
session_id |
event.idm.read_only_udm.network.session_id |
Aus Änderungsprotokoll zugeordnet |
protocol |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
message_code |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
program |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
syslog_prog |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
message_dn |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
message_class |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
syslog_facility_code |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
syslog5424_pri |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
initial_code |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
logstash.ingest.timestamp |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
cisco_timestamp |
event.idm.read_only_udm.additional.fields |
Aus Änderungsprotokoll zugeordnet |
Änderungsprotokoll
Änderungsprotokoll für diesen Parser ansehen
Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten