Cisco Application Centric Infrastructure (ACI)-Logs 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 Parsen von Syslog fehlschlägt, wird davon ausgegangen, dass die Nachricht im JSON-Format vorliegt, und sie wird entsprechend geparst. Anschließend 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 eigenständig und kann unabhängig von Ihren Infrastrukturanforderungen und Logformatpräferenzen implementiert werden.
Option 1: Syslog mit Bindplane-Agent
Mit dieser Option wird die 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öherer oder Linux-Host mit
systemd - Wenn der Agent hinter einem Proxy ausgeführt wird, 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-Agenten bei Google SecOps zu authentifizieren:
- Neue Bereitstellungen: Verwenden Sie die Chronicle API (Option 1).
- Legacy-Bereitstellungen: Verwenden Sie die Authentifizierungsdatei für die Datenaufnahme (Option 2) weiter, 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 -Konsole 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 Editor aus.
- Klicken Sie auf Weiter und dann auf Fertig.
- Klicken Sie auf das von Ihnen erstellte Dienstkonto und rufen Sie 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 und 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 Einstellungen > Datenverwaltung > Erfassungs-Agents auf. Wenn Sie eine reine SIEM-Plattform verwenden, rufen Sie die 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
- Gehen Sie zu Einstellungen > Verwaltung > Profil. Wenn Sie eine reine SIEM-Plattform verwenden, rufen Sie die Einstellungen > Profil auf.
- Kopieren und speichern Sie die Kundennummer aus dem Bereich Organisationsdetails.
Bindplane-Agent installieren
Installieren Sie den Bindplane-Agenten gemäß der folgenden Anleitung auf Ihrem Windows- oder Linux-Betriebssystem.
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 in der Installationsanleitung für den Bindplane-Agenten.
Bindplane-Agent für die Aufnahme von Syslog und das Senden an Google SecOps konfigurieren
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 zu überwachen.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 Sie fürcreds_file_pathden Pfad zur Authentifizierungsdatei für die Datenaufnahme 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.
Starten Sie den Bindplane-Agent neu, 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-collectorDienststatus prüfen:
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.
- Rechtsklicken Sie und wählen Sie Neu starten aus.
Prüfen Sie, ob der Dienst ausgeführt wird:
sc query observiq-otel-collectorPrüfen Sie die Logs auf Fehler:
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
- Drücken Sie
Syslog-Weiterleitung auf Cisco ACI konfigurieren
Out-of-Band-Managementvertrag konfigurieren
- Melden Sie sich in der Cisco APIC-Konsole an.
- Gehen Sie zu Tenants > mgmt > Contracts > Filters.
- Klicken Sie auf Filter erstellen.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
syslog-udp-514ein. - Eintragsname: 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. - In beide Richtungen anwenden: Aktivieren Sie diese Option.
- Name: Geben Sie
- Klicken Sie auf Senden.
- Maximieren Sie den Betreff 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-Monitoring-Zielgruppe 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-Monitoring-Ziel 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
- Gehen Sie zu Fabric > Fabric Policies > Policies > Monitoring > Common Policy.
- Maximieren Sie Callhome/Smart Callhome/SNMP/Syslog/TACACS.
- Rechtsklicken Sie auf Syslog und wählen Sie Syslog-Quelle erstellen aus.
- Geben Sie die folgenden Konfigurationsdetails an:
- Name: Geben Sie
Chronicle-Fabric-Syslogein. - Audit-Logs: Setzen Sie ein Häkchen, um Audit-Ereignisse einzuschließen.
- Ereignisse: Setzen Sie ein Häkchen, um Systemereignisse einzubeziehen.
- Fehler: Aktivieren Sie diese Option, um Fehlerereignisse einzubeziehen.
- Sitzungslogs: Setzen Sie ein Häkchen, um Sitzungslogs einzuschließen.
- Zielgruppe: Wählen Sie
Chronicle-Syslog-Groupaus.
- Name: Geben Sie
- Klicken Sie auf Senden.
Monitoring-Richtlinien konfigurieren – Zugriffsmonitoring
- 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: Setzen Sie ein Häkchen, um Audit-Ereignisse einzuschließen.
- Ereignisse: Setzen Sie ein Häkchen, um Systemereignisse einzubeziehen.
- Fehler: Aktivieren Sie diese Option, um Fehlerereignisse einzubeziehen.
- Sitzungslogs: Setzen Sie ein Häkchen, um Sitzungslogs 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 (Richtlinien für Syslog-Nachrichten).
- 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 mit Google Cloud Storage
Bei dieser Option werden JSON-formatierte Ereignisse, Fehler und Audit-Logs über die APIC REST API 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 Admin > Authentication > AAA (auf älteren Releases).
Erstellen Sie einen 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 die Cloud Run-Funktion erstellen
Die Cloud Run-Funktion benötigt ein Dienstkonto mit Berechtigungen zum Schreiben in den GCS-Bucket und zum Aufrufen durch Pub/Sub.
Dienstkonto erstellen
- Rufen Sie in der GCP Console die Seite IAM und Verwaltung > Dienstkonten auf.
- 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 die Rolle aus.
- Klicken Sie auf Weiter.
- Klicken Sie auf Fertig.
Diese Rollen sind erforderlich für:
- Storage-Objekt-Administrator: Logs in den GCS-Bucket schreiben und Statusdateien verwalten
- Cloud Run-Aufrufer: Pub/Sub erlauben, die Funktion aufzurufen
- Cloud Functions-Invoker: Funktionsaufruf zulassen
IAM-Berechtigungen für GCS-Bucket gewähren
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 Nachrichten veröffentlicht und das die Cloud Run-Funktion abonniert.
- 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 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 (
cisco-aci-collector-sa) aus.
- Dienstkonto: Wählen Sie das Dienstkonto (
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 die Option 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 Wählen Sie das Pub/Sub-Thema ( cisco-aci-trigger) aus.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 * * * |
Batchverarbeitung mit geringem Volumen |
| 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. Achte dabei auf Folgendes:
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: Prüfen Sie die APIC-Anmeldedaten in den Umgebungsvariablen.
- 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 Einstellungen > Datenverwaltung > Feeds auf. Wenn Sie eine reine SIEM-Plattform verwenden, rufen Sie die 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 Logtyp aus.
Klicken Sie auf Dienstkonto abrufen. Eine eindeutige E-Mail-Adresse für das Dienstkonto wird 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 Löschoption aus:
- Nie: Nach Übertragungen werden keine Dateien gelöscht (empfohlen für Tests).
- Übertragene Dateien löschen: Dateien werden nach erfolgreicher Übertragung gelöscht.
- Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach erfolgreicher Übertragung gelöscht.
- 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 Feedkonfiguration 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-Objekt-Betrachter 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 wird aus dem Rohlogfeld „aci_tag“ übernommen. |
| 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 umgewandelt. |
| Beschreibung | read_only_udm.security_result.description | Der Wert wird aus dem Feld „description“ des Rohlogs ü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“ 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 wird aus dem Rohprotokollfeld „lifecycle_state“ übernommen. |
| 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 ist 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 Rohprotokollfeld „message_type“ übernommen, nachdem die eckigen Klammern entfernt wurden. |
| node_link | read_only_udm.principal.process.file.full_path | Der Wert wird aus dem Rohlogfeld „node_link“ übernommen. |
| 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 konvertiert 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 umgewandelt. |
| 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 Rohlogfeld „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 -> INFORMATION; 3, 4 -> MITTEL; 0, 1, 2 -> HOCH. |
| 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“ vorhanden ist und „Proto“ vorhanden ist, wird „NETWORK_CONNECTION“ festgelegt. 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