Apache Kafka in Google SecOps einbinden

In diesem Dokument wird beschrieben, wie Sie Apache Kafka in Google Security Operations (Google SecOps) einbinden.

Hinweis :Alle Tests für Version 1 dieser Integration wurden mit Apache Kafka Version 4.0.0 durchgeführt und validiert.

Anwendungsfälle

Die Apache Kafka-Integration kann für die folgenden Anwendungsfälle verwendet werden:

• Echtzeit-Aufnahme von Sicherheitslogs: Sicherheitsereignisse aus Kafka-Themen werden automatisch in Google SecOps aufgenommen und verarbeitet. Dies ermöglicht eine zentrale Logverwaltung und Echtzeitanalyse, um Benachrichtigungen auf Grundlage von Streamingdaten zu generieren.

• Ereignisgesteuerte Automatisierung: Automatisierte Playbooks in Google SecOps werden durch bestimmte Sicherheitsereignisse oder Nachrichten ausgelöst, die aus einem Kafka-Thema gestreamt werden. Dadurch wird die Reaktion auf kritische Ereignisse wie die Anmeldung eines Nutzers von einem ungewöhnlichen Standort beschleunigt.

• Anreicherung mit Threat Intelligence: Benutzerdefinierte Threat Intelligence-Feeds aus Kafka-Themen abrufen, um vorhandene Benachrichtigungen und Fälle anzureichern. So erhalten Analysten aktuellen Kontext zu Indicators of Compromise (IOCs) und die Genauigkeit der Bedrohungsanalyse wird verbessert.

Hinweis

Bevor Sie die Apache Kafka-Integration in Google SecOps konfigurieren, müssen Sie die folgenden Voraussetzungen erfüllen:

• Apache Kafka-Server: Sie benötigen Zugriff auf einen laufenden Apache Kafka-Server mit den erforderlichen Kafka-Brokern und -Themen.

• Docker-Image für Remote-Agent: Beim Erstellen von Remote-Agents müssen Sie ein Debian-basiertes Image verwenden. Verwenden Sie das folgende Bild, um die Kompatibilität zu prüfen:

  us-docker.pkg.dev/siem-ar-public/images/agent-debian:latest
  

Integrationsparameter

Für die Apache Kafka-Integration sind die folgenden Parameter erforderlich:

Parameter	Beschreibung
Kafka brokers	Erforderlich. Eine durch Kommas getrennte Liste von Kafka-Brokern, mit denen eine Verbindung hergestellt werden soll, im Format hostname:port.
Use TLS for connection	Optional. Wenn diese Option ausgewählt ist, verwendet die Integration die TLS-Verschlüsselung für die Authentifizierung. Für diesen Parameter ist ein Zertifizierungsstellenzertifikat erforderlich. Standardmäßig nicht aktiviert.
Use SASL PLAIN with TLS for connection	Optional. Wenn diese Option ausgewählt ist, verwendet die Integration den SASL PLAIN-Mechanismus für Nutzernamen und Passwörter zur Authentifizierung. Diese Option wird nur mit TLS-Verschlüsselung unterstützt und erfordert sowohl einen SASL-Nutzernamen und ein SASL-Passwort als auch ein Zertifizierungsstellenzertifikat. Standardmäßig nicht aktiviert.
CA certificate of Kafka server	Optional. Das Zertifizierungsstellenzertifikat, das zum Überprüfen der Identität des Kafka-Servers verwendet wird. Dieser Parameter ist erforderlich, wenn SASL aktiviert ist.
Client certificate	Optional. Das Clientzertifikat für die gegenseitige TLS-Authentifizierung mit dem Kafka-Server. Dieser Parameter ist erforderlich, wenn gegenseitiges TLS (mTLS) aktiviert ist.
Client certificate key	Optional. Der private Schlüssel, der dem Zertifikat des Clients entspricht und für die gegenseitige TLS-Authentifizierung verwendet wird. Dieser Parameter ist erforderlich, wenn gegenseitiges TLS (mTLS) aktiviert ist.
Client certificate key password	Optional. Das Passwort, das zum Entschlüsseln des privaten Schlüssels des Clientzertifikats verwendet wird. Dieser Parameter ist erforderlich, wenn gegenseitiges TLS (mTLS) aktiviert ist.
SASL PLAIN Username	Optional. Der Nutzername für die SASL PLAIN-Authentifizierung bei Kafka-Brokern. Dieser Parameter ist erforderlich, wenn SASL aktiviert ist.
SASL PLAIN Password	Optional. Das Passwort für die SASL PLAIN-Authentifizierung bei Kafka-Brokern. Dieser Parameter ist erforderlich, wenn SASL aktiviert ist.

Eine Anleitung zum Konfigurieren einer Integration in Google SecOps finden Sie unter Integrationen konfigurieren.

Bei Bedarf können Sie später Änderungen vornehmen. Nachdem Sie eine Integrationsinstanz konfiguriert haben, können Sie sie in Playbooks verwenden. Weitere Informationen zum Konfigurieren und Unterstützen mehrerer Instanzen finden Sie unter Mehrere Instanzen unterstützen.

Aktionen

Weitere Informationen zu Aktionen finden Sie unter Ausstehende Aktionen in „Mein Arbeitsbereich“ beantworten und Manuelle Aktion ausführen.

Ping

Verwenden Sie die Aktion Ping, um die Verbindung zu Apache Kafka zu testen.

Diese Aktion wird nicht für Google SecOps-Elemente ausgeführt.

Aktionseingaben

Keine.

Aktionsausgaben

Die Aktion Ping bietet die folgenden Ausgaben:

Ausgabetyp der Aktion	Verfügbarkeit
Anhang im Fall-Repository	Nicht verfügbar
Link zum Fall‑Repository	Nicht verfügbar
Tabelle „Fall-Repository“	Nicht verfügbar
Anreicherungstabelle	Nicht verfügbar
JSON-Ergebnis	Nicht verfügbar
Ausgabenachrichten	Verfügbar
Scriptergebnis.	Verfügbar

Ausgabenachrichten

Die Aktion Ping kann die folgenden Ausgabenachrichten zurückgeben:

Ausgabemeldung	Nachrichtenbeschreibung
Successfully connected to the Apache Kafka server with the provided connection parameters!	Die Aktion wurde ausgeführt.
Failed to connect to the Apache Kafka server! Error is ERROR_REASON	Die Aktion ist fehlgeschlagen. Überprüfen Sie die Verbindung zum Server, die Eingabeparameter oder die Anmeldedaten.

Scriptergebnis

In der folgenden Tabelle ist der Wert für die Ausgabe des Skriptergebnisses bei Verwendung der Aktion Ping aufgeführt:

Name des Scriptergebnisses	Wert
is_success	True oder False

Connectors

Weitere Informationen zum Konfigurieren von Connectors in Google SecOps finden Sie unter Daten aufnehmen (Connectors).

Hinweis:Um Datenverlust zu vermeiden, wird in Connectors die Ereignisabflachung verwendet. Wenn eine Rohbenachrichtigung eine Liste von Entitäten enthält, z. B. mehrere E-Mail-Adressen, Hostnamen oder IP-Adressen, werden diese von Connectors automatisch in separate, eindeutige Ereignisse umgewandelt.

Wenn beispielsweise eine einzelne Rohbenachrichtigung drei verschiedene E‑Mail-Adressen enthält, wird sie als drei separate Ereignisse aufgenommen, die jeweils eine eigene E‑Mail-Adresse enthalten.

So wird sichergestellt, dass jede Einheit als eindeutiges Asset richtig indexiert wird und in Playbooks vollständig durchsuchbar und nutzbar ist.

Apache Kafka – Messages Connector

Mit dem Apache Kafka – Messages Connector können Sie Nachrichten aus Apache Kafka abrufen.

Der Connector ruft Nachrichten aus einem angegebenen Kafka-Thema ab und kann sie je nach Nachrichtenformat unterschiedlich verarbeiten. Wenn eine Nachricht ein gültiges JSON-Objekt ist, extrahiert der Connector bestimmte Felder für die Erstellung und Zuordnung von Benachrichtigungen. Wenn die Nachricht ein einfacher String ist, werden die Daten als Rohereignisdaten erfasst.

Der Connector übernimmt die Zuordnung von Schweregraden, die Erstellung von Vorlagen für Warnungsnamen und die Generierung eindeutiger IDs basierend auf den von Ihnen angegebenen Parametern.

JSON-Schweregradzuordnung

Um den Schweregrad der Benachrichtigung zuzuordnen, müssen Sie angeben, welches Feld der Apache Kafka – Messages Connector verwendet, um den Wert für den Schweregrad im Parameter Severity Mapping JSON abzurufen. Die Connector-Antwort kann Werttypen wie integer, float und string enthalten.

Der Apache Kafka – Messages Connector liest die Werte integer und float und ordnet sie gemäß den Google SecOps-Einstellungen zu. In der folgenden Tabelle sehen Sie die Zuordnung der integer-Werte zum Schweregrad in Google SecOps:

Ganzzahlwert	Zugeordneter Schweregrad
100	Critical
Von 80 bis 100	High
Von 60 bis 80	Medium
Von 40 bis 60	Low
Weniger als 40	Informational

Wenn die Antwort den Wert string enthält, muss der Pub/Sub – Messages Connector zusätzlich konfiguriert werden.

Anfangs wird der Standardwert so angezeigt:

{
    "Default": 60
}

Wenn sich die für die Zuordnung erforderlichen Werte im JSON-Schlüssel event_severity befinden, können die Werte so aussehen:

• "Malicious"
• "Benign"
• "Unknown"

Um die event_severity-JSON-Schlüsselwerte zu parsen und dafür zu sorgen, dass das JSON-Objekt das richtige Format hat, konfigurieren Sie den Parameter Severity Mapping JSON so:

{
    "event_severity": {
        "Malicious": 100,
        "Unknown": 60,
        "Benign": -1
    },
    "Default": 50
}

Der "Default"-Wert ist erforderlich.

Wenn es mehrere Übereinstimmungen für dasselbe JSON-Objekt gibt, priorisiert der Apache Kafka – Messages Connector den ersten JSON-Objektschlüssel.

Wenn Sie mit Feldern arbeiten möchten, die integer- oder float-Werte enthalten, konfigurieren Sie den Schlüssel und einen leeren String im Parameter Severity Mapping JSON:

{
  "Default":"60",
  "integer_field": "",
  "float_field": ""
}

Connector-Eingaben

Für den Apache Kafka – Messages Connector sind die folgenden Parameter erforderlich:

Parameter	Beschreibung
Product Field Name	Erforderlich. Der Name des Felds, in dem der Produktname gespeichert ist. Der Produktname wirkt sich hauptsächlich auf die Zuordnung aus. Um den Zuordnungsprozess für den Connector zu optimieren und zu verbessern, wird der Standardwert in einen Fallback-Wert aufgelöst, auf den im Code verwiesen wird. Ungültige Eingaben für diesen Parameter werden standardmäßig in einen Fallback-Wert aufgelöst. Der Standardwert ist Product Name.
Event Field Name	Erforderlich. Der Name des Felds, das den Ereignisnamen (Untertyp) bestimmt. Der Standardwert ist event_type.
Environment Field Name	Optional. Der Name des Felds, in dem der Name der Umgebung gespeichert ist. Wenn das Feld „environment“ fehlt, wird der Standardwert verwendet. Der Standardwert ist "".
Environment Regex Pattern	Optional. Ein Muster für einen regulären Ausdruck, der für den Wert im Feld Environment Field Name ausgeführt werden soll. Mit diesem Parameter können Sie das Feld „environment“ mithilfe der Logik für reguläre Ausdrücke bearbeiten. Verwenden Sie den Standardwert .*, um den erforderlichen Rohwert Environment Field Name abzurufen. Wenn das Muster des regulären Ausdrucks null oder leer ist oder der Umgebungswert null ist, ist das endgültige Umgebungsergebnis die Standardumgebung.
Script Timeout (Seconds)	Erforderlich. Das Zeitlimit in Sekunden für den Python-Prozess, in dem das aktuelle Script ausgeführt wird. Der Standardwert ist 180.
Kafka brokers	Erforderlich. Eine durch Kommas getrennte Liste von Kafka-Brokern, mit denen eine Verbindung hergestellt werden soll, im Format hostname:port.
Use TLS for connection	Optional. Wenn diese Option ausgewählt ist, verwendet die Integration die TLS-Verschlüsselung für die Authentifizierung. Für diesen Parameter ist ein CA-Zertifikat erforderlich. Standardmäßig nicht aktiviert.
Use SASL PLAIN with TLS for connection	Optional. Wenn diese Option ausgewählt ist, verwendet die Integration den SASL PLAIN-Mechanismus für Nutzernamen und Passwörter zur Authentifizierung. Für diese Option müssen ein SASL-Nutzername und ein SASL-Passwort angegeben werden. Sie wird nur mit TLS-Verschlüsselung unterstützt, für die ein CA-Zertifikat erforderlich ist. Standardmäßig nicht aktiviert.
CA certificate of Kafka server	Optional. Das Zertifizierungsstellenzertifikat, das zum Überprüfen der Identität des Kafka-Servers verwendet wird.
Client certificate	Optional. Das Clientzertifikat für die gegenseitige TLS-Authentifizierung mit dem Kafka-Server.
Client certificate key	Optional. Der private Schlüssel, der dem Zertifikat des Clients entspricht und für die gegenseitige TLS-Authentifizierung verwendet wird.
Client certificate key password	Optional. Das Passwort, das zum Entschlüsseln des privaten Schlüssels des Clientzertifikats verwendet wird.
SASL PLAIN Username	Optional. Der Nutzername für die SASL PLAIN-Authentifizierung bei Kafka-Brokern.
SASL PLAIN Password	Optional. Das Passwort für die SASL PLAIN-Authentifizierung bei Kafka-Brokern.
Topic	Erforderlich. Das Kafka-Thema, aus dem Vorfälle abgerufen werden.
Consumer Group ID	Optional. Die Kennung der Verbrauchergruppe, die beim Abrufen von Vorfällen verwendet wird. Wenn kein Wert angegeben ist, wird eine eindeutige ID generiert.
Partitions	Optional. Eine CSV-Liste der Partitionen, aus denen Nachrichten abgerufen werden sollen.
Initial Offset	Optional. Gibt an, ab wo der Connector Nachrichten aus einer Kafka-Partition abruft. Sie können eine positive Ganzzahl angeben, um an einem bestimmten Offset zu beginnen, oder die Werte earliest oder latest verwenden, um mit dem Abrufen am Anfang oder Ende der Partition zu beginnen.
Poll Timeout	Optional. Ein Poll-Zeitlimit zum Empfangen einer Nachricht von Kafka in Sekunden.
Case Name Template	Optional. Eine Vorlage zum Definieren eines benutzerdefinierten Fallnamens. Der Connector fügt dem Ereignis einen custom_case_name-Schlüssel hinzu. Sie können Platzhalter im Format FIELD_NAME verwenden, die mit den String-Werten des ersten Ereignisses gefüllt werden. Beispiel: Phishing - EVENT_MAILBOX.
Alert Name Template	Erforderlich. Eine Vorlage zum Definieren des Benachrichtigungsnamens. Sie können Platzhalter im Format FIELD_NAME verwenden, die mit den String-Werten des ersten Ereignisses gefüllt werden. Beispiel: Phishing - EVENT_MAILBOX. Wenn kein Wert angegeben wird oder die Vorlage ungültig ist, verwendet der Connector einen Standardnamen für die Benachrichtigung.
Rule Generator Template	Erforderlich. Eine Vorlage zum Definieren des Regelgenerators. Sie können Platzhalter im Format FIELD_NAME verwenden, die mit den String-Werten des ersten Ereignisses gefüllt werden. Beispiel: Phishing - EVENT_MAILBOX. Wenn kein Wert angegeben wird oder die Vorlage ungültig ist, verwendet der Connector einen Standardnamen für die Regelgenerierung.
Timestamp Field	Erforderlich. Der Feldname in der Kafka-Nachricht, der den Zeitstempel der Google SecOps-Benachrichtigung enthält. Wenn der Zeitstempel nicht im Unix-Epochenformat vorliegt, muss sein Format im Parameter Timestamp Format definiert werden.
Timestamp Format	Optional. Das Format des Zeitstempels der Nachricht, das für Zeitstempel erforderlich ist, die nicht auf der Unix-Epoche basieren. Verwenden Sie die Standard-Python-Formatcodes strftime. Wenn der Zeitstempel nicht im Unix-Epochenformat vorliegt und dieser Parameter nicht konfiguriert ist, schlägt der Connector fehl.
Severity Mapping JSON	Erforderlich. Das JSON-Objekt, das vom Connector verwendet wird, um den Schweregrad aus der Nachricht zu extrahieren und der Google SecOps-Prioritätsskala zuzuordnen. Der Standardwert ist {"Default": "60"}.
Unique ID Field	Optional. Der Name des Felds, das als eindeutige Nachrichten-ID verwendet werden soll. Wenn kein Wert angegeben ist, generiert und verwendet der Connector einen SHA-256-Hash des Nachrichteninhalts als Nachrichten-ID.
Max Messages To Fetch	Erforderlich. Die maximale Anzahl von Nachrichten, die der Connector für jede Iteration verarbeitet. Der Standardwert ist 100.
Disable Overflow	Optional. Wenn diese Option ausgewählt ist, ignoriert der Connector den Google SecOps-Überlaufmechanismus. Standardmäßig aktiviert.
Verify SSL	Erforderlich. Wenn diese Option ausgewählt ist, wird das SSL-Zertifikat bei der Verbindung zum Apache Kafka-Server validiert. Standardmäßig aktiviert.
Proxy Server Address	Optional. Die Adresse des zu verwendenden Proxyservers.
Proxy Username	Optional. Der Proxy-Nutzername für die Authentifizierung.
Proxy Password	Optional. Das Proxy-Passwort für die Authentifizierung.

Connector-Regeln

Der Connector unterstützt Proxys.

Connector-Benachrichtigungen

In der folgenden Tabelle wird die Zuordnung von Apache Kafka-Nachrichtenfeldern zu Google SecOps-Benachrichtigungsfeldern beschrieben:

Siemplify-Benachrichtigungsfeld	Apache Kafka-Nachrichtenfeld
SourceSystemName	Wird vom Framework ausgefüllt.
TicketId	Der Wert des Felds „Eindeutige ID“ oder ein SHA-256-Hash der Nachricht.
DisplayId	ApacheKafka_{unique id or hash}_{connector identifier}
Name	Der von Alert Name Template generierte Wert.
Reason	–
Description	–
DeviceVendor	Fest codiert: Apache Kafka
DeviceProduct	Fallback-Wert: Message
Priority	Wird aus dem Parameter Severity Mapping JSON abgeleitet.
RuleGenerator	Der von Rule Generator Template generierte Wert.
SourceGroupingIdentifier	–
StartTime	Konvertiert aus dem Timestamp Field.
EndTime	Konvertiert aus dem Timestamp Field.
Siemplify Alert - Extensions	–
Siemplify Alert - Attachments	–

Connector-Ereignisse

Hier ist ein Beispiel für ein Connector-Ereignis:

{
  "notificationConfigName": "organizations/ORGANIZATION_ID/notificationConfigs/soar_connector_CONNECTOR_ID_toxic_notifications_config",
  "finding": {
    "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID",
    "parent": "organizations/ORGANIZATION_ID/sources/SOURCE_ID",
    "resourceName": "//compute.googleapis.com/projects/PROJECT_ID/global/firewalls/FIREWALL_ID",
    "state": "ACTIVE",
    "category": "OPEN_NETBIOS_PORT",
    "externalUri": "https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
    "sourceProperties": {
      "Recommendation": "Restrict the firewall rules at: https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
      "ExceptionInstructions": "Add the security mark \"allow_open_netbios_port\" to the asset with a value of \"true\" to prevent this finding from being activated again.",
      "Explanation": "Firewall rules that allow connections from all IP addresses on TCP ports 137-139 or UDP ports 137-139 may expose NetBIOS services to attackers.",
      "ScannerName": "FIREWALL_SCANNER",
      "ResourcePath": [
        "projects/PROJECT_ID/",
        "folders/FOLDER_ID_1/",
        "folders/FOLDER_ID_2/",
        "organizations/ORGANIZATION_ID/"
      ],
      "ExposedService": "NetBIOS",
      "OpenPorts": {
        "TCP": [
          137.0,
          138.0,
          139.0
        ],
        "UDP": [
          137.0,
          138.0,
          139.0
        ]
      },
      "compliance_standards": {
        "iso": [
          {
            "ids": [
              "A.13.1.1"
            ]
          }
        ],
        "pci": [
          {
            "ids": [
              "1.2.1"
            ]
          }
        ],
        "nist": [
          {
            "ids": [
              "SC-7"
            ]
          }
        ]
      },
      "ReactivationCount": 4.0
    },
    "securityMarks": {
      "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID/securityMarks",
      "marks": {
        "USER_ID": "SECURITY_MARK"
      }
    },
    "eventTime": "2024-08-30T14:44:37.973090Z",
    "createTime": "2024-06-24T07:08:54.777Z",
    "propertyDataTypes": {
      "ResourcePath": {
        "listValues": {
          "propertyDataTypes": [
            {
              "primitiveDataType": "STRING"
            }
          ]
        }
      },
      "ReactivationCount": {
        "primitiveDataType": "NUMBER"
      },
      "Explanation": {
        "primitiveDataType": "STRING"
      },
      "ExposedService": {
        "primitiveDataType": "STRING"
      },
      "ScannerName": {
        "primitiveDataType": "STRING"
      }
    }
  }
}

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten