Elasticsearch-Logs erfassen

Parserversion: 1.0

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Elasticsearch-Logs mit Amazon S3 in Google Security Operations aufnehmen. Der Parser wandelt Rohlogs im JSON-Format in ein einheitliches Datenmodell (Unified Data Model, UDM) um. Es werden Felder aus verschachtelten JSON-Strukturen extrahiert, UDM-Feldern zugeordnet und die Daten werden mit sicherheitsrelevanten Kontextinformationen wie Schweregraden und Nutzerrollen angereichert.

Hinweis

  • Eine Google SecOps-Instanz
  • Privilegierter Zugriff auf die Elasticsearch-Clusteradministration
  • Privilegierter Zugriff auf AWS (S3, IAM, EC2)
  • EC2-Instanz oder persistenter Host zum Ausführen von Logstash

Elasticsearch-Voraussetzungen

  1. Melden Sie sich als Administrator in Ihrem Elasticsearch-Cluster an.
  2. Prüfen Sie, ob Ihr Elasticsearch-Abo Sicherheitsfunktionen enthält (für die Überwachungsprotokollierung erforderlich).
  3. Notieren Sie sich den Namen und die Version Ihres Elasticsearch-Clusters.
  4. Geben Sie den Pfad an, in den Audit-Logs geschrieben werden sollen (Standard: $ES_HOME/logs/<clustername>_audit.json).

Elasticsearch-Audit-Logging aktivieren

  1. Bearbeiten Sie auf jedem Elasticsearch-Knoten die Konfigurationsdatei elasticsearch.yml.

  2. Fügen Sie die folgende Einstellung hinzu:

    xpack.security.audit.enabled: true

  3. Führen Sie einen fortlaufenden Neustart des Clusters aus, um die Änderungen zu übernehmen:

    • Shard-Zuweisung deaktivieren: PUT _cluster/settings {"persistent": {"cluster.routing.allocation.enable": "primaries"}}
    • Halten Sie jeden Knoten einzeln an und starten Sie ihn neu.
    • Shard-Zuweisung wieder aktivieren: PUT _cluster/settings {"persistent": {"cluster.routing.allocation.enable": null}}

  4. Prüfen Sie, ob Audit-Logs im Logverzeichnis unter <clustername>_audit.json generiert werden.

AWS S3-Bucket und IAM für Google SecOps konfigurieren

  1. Erstellen Sie einen Amazon S3-Bucket. Folgen Sie dazu der Anleitung in diesem Nutzerhandbuch: Bucket erstellen.
  2. Speichern Sie den Namen und die Region des Buckets zur späteren Verwendung (z. B. elastic-search-logs).
  3. Erstellen Sie einen Nutzer gemäß dieser Anleitung: IAM-Nutzer erstellen.
  4. Wählen Sie den erstellten Nutzer aus.
  5. Wählen Sie den Tab Sicherheitsanmeldedaten aus.
  6. Klicken Sie im Bereich Zugriffsschlüssel auf Zugriffsschlüssel erstellen.
  7. Wählen Sie Drittanbieterdienst als Anwendungsfall aus.
  8. Klicken Sie auf Weiter.
  9. Optional: Fügen Sie ein Beschreibungstag hinzu.
  10. Klicken Sie auf Zugriffsschlüssel erstellen.
  11. Klicken Sie auf CSV-Datei herunterladen, um den Access Key (Zugriffsschlüssel) und den Secret Access Key (geheimer Zugriffsschlüssel) für die zukünftige Verwendung zu speichern.
  12. Klicken Sie auf Fertig.
  13. Wählen Sie den Tab Berechtigungen aus.
  14. Klicken Sie im Bereich Berechtigungsrichtlinien auf Berechtigungen hinzufügen.
  15. Wählen Sie Berechtigungen hinzufügen aus.
  16. Wählen Sie Richtlinien direkt anhängen aus.
  17. Suchen Sie nach der Richtlinie AmazonS3FullAccess.
  18. Wählen Sie die Richtlinie aus.
  19. Klicken Sie auf Weiter.
  20. Klicken Sie auf Berechtigungen hinzufügen.

Logstash so konfigurieren, dass Audit-Logs an S3 gesendet werden

  1. Installieren Sie Logstash auf einer EC2-Instanz oder einem persistenten Host, der auf die Elasticsearch-Audit-Log-Dateien zugreifen kann.

  2. Installieren Sie das S3-Ausgabe-Plug-in, falls es noch nicht vorhanden ist:

    bin/logstash-plugin install logstash-output-s3

  3. Erstellen Sie eine Logstash-Konfigurationsdatei (elastic-to-s3.conf):

    input {
  file {
    path => "/path/to/elasticsearch/logs/*_audit.json"
    start_position => "beginning"
    codec => "json"              # audit file: 1 JSON object per line
    sincedb_path => "/var/lib/logstash/sincedb_elastic_search"
    exclude => ["*.gz"]
  }
}

filter {
  # Intentionally minimal: do NOT reshape audit JSON the ELASTIC_SEARCH parser expects.
  # If you must add metadata for ops, put it under [@metadata] so it won't be written.
  # ruby { code => "event.set('[@metadata][ingested_at]', Time.now.utc.iso8601)" }
}

output {
  s3 {
    access_key_id => "YOUR_AWS_ACCESS_KEY"
    secret_access_key => "YOUR_AWS_SECRET_KEY"
    region => "us-east-1"
    bucket => "elastic-search-logs"
    prefix => "logs/%{+YYYY}/%{+MM}/%{+dd}/"

    codec => "json_lines"        # NDJSON output (1 JSON object per line)
    encoding => "gzip"           # compress objects

    server_side_encryption => true
    # Optionally for KMS:
    # server_side_encryption_kms_key_id => "arn:aws:kms:REGION:ACCT:key/KEY_ID"

    size_file => 104857600       # 100MB rotation
    time_file => 300             # 5 min rotation
  }
}

  4. Starten Sie Logstash mit der Konfiguration:

    bin/logstash -f elastic-to-s3.conf

Optional: IAM-Nutzer mit Lesezugriff für Google SecOps erstellen

  1. Rufen Sie die AWS-Konsole > IAM > Nutzer > Nutzer hinzufügen auf.
  2. Klicken Sie auf Nutzer hinzufügen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Nutzer: Geben Sie secops-reader ein.
    • Zugriffstyp: Wählen Sie Zugriffsschlüssel – programmatischer Zugriff aus.
  4. Klicken Sie auf Nutzer erstellen.
  5. Minimale Leserichtlinien (benutzerdefiniert) anhängen: Nutzer > secops-reader > Berechtigungen > Berechtigungen hinzufügen > Richtlinien direkt anhängen > Richtlinie erstellen.

  6. Geben Sie im JSON-Editor die folgende Richtlinie ein:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::elastic-search-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::elastic-search-logs"
    }
  ]
}

  7. Legen Sie secops-reader-policy als Name fest.

  8. Gehen Sie zu Richtlinie erstellen> suchen/auswählen > „Weiter“ > Berechtigungen hinzufügen.

  9. Gehen Sie zu Sicherheitsanmeldedaten > Zugriffsschlüssel > Zugriffsschlüssel erstellen.

  10. Laden Sie die CSV herunter. Diese Werte werden in den Feed eingegeben.

Feed in Google SecOps konfigurieren, um Elasticsearch-Logs aufzunehmen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf + Neuen Feed hinzufügen.
  3. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. Elasticsearch Logs.
  4. Wählen Sie Amazon S3 V2 als Quelltyp aus.
  5. Wählen Sie Elastic Search als Protokolltyp aus.
  6. Klicken Sie auf Weiter.
  7. Geben Sie Werte für die folgenden Eingabeparameter an:
    • S3-URI: s3://elastic-search-logs/logs/
    • Optionen zum Löschen von Quellen: Wählen Sie die gewünschte Option zum Löschen aus.
    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.
    • Zugriffsschlüssel-ID: Nutzerzugriffsschlüssel mit Zugriff auf den S3-Bucket.
    • Geheimer Zugriffsschlüssel: Der geheime Schlüssel des Nutzers mit Zugriff auf den S3-Bucket.
    • Asset-Namespace: Der Asset-Namespace.
    • Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet wird.
  8. Klicken Sie auf Weiter.
  9. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
Level security_result.severity Die Logik prüft den Wert des Felds „Level“ und ordnet ihn der entsprechenden UDM-Schweregradstufe zu:
 – „INFO“, „ALL“, „OFF“, „TRACE“, „DEBUG“ werden „INFORMATIONAL“ zugeordnet.
 – „WARN“ wird „LOW“ zugeordnet.
 – „ERROR“ wird „ERROR“ zugeordnet.
 – „FATAL“ wird „CRITICAL“ zugeordnet.
message.@timestamp timestamp Der Zeitstempel wird aus dem Feld „@timestamp“ im Feld „message“ des Rohlogs im Format „JJJJ-MM-TTTHH:mm:ss.SSS“ geparst.
message.action security_result.action_details Der Wert wird aus dem Feld „action“ im Feld „message“ des Rohlogs übernommen.
message.event.action security_result.summary Der Wert wird aus dem Feld „event.action“ im Feld „message“ des Rohlogs übernommen.
message.event.type metadata.product_event_type Der Wert wird aus dem Feld „event.type“ im Feld „message“ des Rohlogs übernommen.
message.host.ip target.ip Der Wert wird aus dem Feld „host.ip“ im Feld „message“ des Rohlogs übernommen.
message.host.name target.hostname Der Wert wird aus dem Feld „host.name“ im Feld „message“ des Rohlogs übernommen.
message.indices target.labels.value Der Wert wird aus dem Feld „indices“ im Feld „message“ des Rohlogs übernommen.
message.mrId target.hostname Der Wert wird aus dem Feld „mrId“ im Feld „message“ des Rohlogs übernommen.
message.node.id principal.asset.product_object_id Der Wert stammt aus dem Feld „node.id“ im Feld „message“ des Rohlogs.
message.node.name target.asset.hostname Der Wert wird aus dem Feld „node.name“ im Feld „message“ des Rohlogs übernommen.
message.origin.address principal.ip Die IP-Adresse wird aus dem Feld „origin.address“ im Feld „message“ des Rohlogs extrahiert, indem die Portnummer entfernt wird.
message.origin.type principal.resource.resource_subtype Der Wert stammt aus dem Feld „origin.type“ im Feld „message“ des Rohlogs.
message.properties.host_group principal.hostname Der Wert wird aus dem Feld „properties.host_group“ im Feld „message“ des Rohlogs übernommen.
message.properties.host_group target.group.group_display_name Der Wert wird aus dem Feld „properties.host_group“ im Feld „message“ des Rohlogs übernommen.
message.request.id target.resource.product_object_id Der Wert wird aus dem Feld „request.id“ im Feld „message“ des Rohlogs übernommen.
message.request.name target.resource.name Der Wert wird aus dem Feld „request.name“ im Feld „message“ des Rohlogs übernommen.
message.user.name principal.user.userid Der Wert wird aus dem Feld „user.name“ im Feld „message“ des Rohlogs übernommen.
message.user.realm principal.user.attribute.permissions.name Der Wert wird aus dem Feld „user.realm“ im Feld „message“ des Rohlogs übernommen.
message.user.roles about.user.attribute.roles.name Der Wert wird aus dem Feld „user.roles“ im Feld „message“ des Rohlogs übernommen.
metadata.event_type Fest codierter Wert: „USER_RESOURCE_ACCESS“
metadata.log_type Fest codierter Wert: „ELASTIC_SEARCH“
metadata.product_name Fest codierter Wert: „ELASTICSEARCH“
metadata.vendor_name Fest codierter Wert: „ELASTIC“
principal.port Die Portnummer wird aus dem Feld „origin.address“ im Feld „message“ des Rohlogs extrahiert.
target.labels.key Fest codierter Wert: „Indice“

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

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