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 unformatierte JSON-formatierte Logs in ein einheitliches Datenmodell (UDM) um. Sie extrahiert Felder aus verschachtelten JSON-Strukturen, ordnet sie UDM-Feldern zu und reichert die Daten mit sicherheitsrelevantem Kontext wie Schweregraden und Nutzerrollen an.

Hinweis

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

Voraussetzungen für Elasticsearch abrufen

  1. Melden Sie sich als Administrator in Ihrem Elasticsearch-Cluster an.
  2. Prüfen Sie, ob Ihr Elasticsearch-Abo Sicherheitsfunktionen enthält (für das Audit-Logging 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 (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 rollierenden Neustart des Clusters durch, um die Änderungen zu übernehmen:

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

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

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

  1. Erstellen Sie einen Amazon S3-Bucket gemäß 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äß diesem Nutzerhandbuch: IAM-Nutzer erstellen.
  4. Wählen Sie den erstellten Nutzer aus.
  5. Wählen Sie den Tab Sicherheitsanmeldedaten aus.
  6. Klicken Sie im Bereich Access Keys (Zugriffsschlüssel) auf Create Access Key (Zugriffsschlüssel erstellen).
  7. Wählen Sie Drittanbieterdienst als Anwendungsfall aus.
  8. Klicken Sie auf Weiter.
  9. Optional: Fügen Sie ein Beschreibungs-Tag hinzu.
  10. Klicken Sie auf Zugriffsschlüssel erstellen.
  11. Klicken Sie auf CSV-Datei herunterladen, um den Zugriffsschlüssel und den geheimen Zugriffsschlüssel zur späteren 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 für das Senden von Audit-Logs an S3 konfigurieren

  1. Installieren Sie Logstash auf einer EC2-Instanz oder einem nichtflüchtigen Host, der auf die Elasticsearch-Audit-Logdateien 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 > Users > Add users 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. Richtlinie für minimalen Lesezugriff (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 Einstellungen > Datenverwaltung > Feeds auf. Wenn Sie eine reine SIEM-Plattform verwenden, rufen Sie die 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 als Quelltyp Amazon S3 V2 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 der Quelle: Wählen Sie die gewünschte Löschoption aus.
    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.
    • Zugriffsschlüssel-ID: Der 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 dem entsprechenden UDM-Schweregrad zu:
– „INFO“, „ALL“, „OFF“, „TRACE“, „DEBUG“ werden „INFORMATIONAL“ zugeordnet.
: „WARN“ wird „LOW“ zugeordnet.
: „ERROR“ ist „ERROR“ zugeordnet.
: „FATAL“ wird „CRITICAL“ zugeordnet.
message.@timestamp timestamp Der Zeitstempel wird aus dem Feld „@timestamp“ im Feld „message“ des Rohlogs im Format „yyyy-MM-ddTHH: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 wird aus dem Feld „node.id“ im Feld „message“ des Rohlogs übernommen.
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 wird aus dem Feld „origin.type“ im Feld „message“ des Rohlogs übernommen.
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 Hartcodierter Wert: „USER_RESOURCE_ACCESS“
metadata.log_type Hartcodierter Wert: „ELASTIC_SEARCH“
metadata.product_name Hartcodierter Wert: „ELASTICSEARCH“
metadata.vendor_name Hartcodierter Wert: „ELASTIC“
principal.port Die Portnummer wird aus dem Feld „origin.address“ im Feld „message“ des Rohlogs extrahiert.
target.labels.key Hartcodierter Wert: „Indice“

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

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