Collecter les journaux Elastic Defend

Compatible avec :

Ce document explique comment ingérer des journaux Elastic Defend dans Google Security Operations à l'aide de Google Cloud Storage V2 avec une fonction Cloud Run.

Elastic Defend est une solution de détection et de réponse au niveau des points de terminaison (EDR) d'Elastic Security. Elle offre des fonctionnalités de prévention, de détection et de réponse, ainsi qu'une visibilité approfondie sur les systèmes d'exploitation Windows, macOS et Linux. Il surveille l'exécution des processus, l'activité des fichiers, les connexions réseau, les modifications du registre et les chargements de bibliothèque pour détecter et prévenir les menaces au niveau du point de terminaison. Les données sont stockées dans Elasticsearch et peuvent être récupérées à l'aide de l'API Elasticsearch Search.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps
  • Un projet Google Cloud avec les API suivantes activées :
    • Cloud Storage
    • Cloud Run Functions
    • Cloud Scheduler
    • Pub/Sub
    • IAM
  • Accès à un cluster Elasticsearch avec Elastic Defend déployé
  • Autorisations permettant de créer des clés API dans Elasticsearch (privilège de cluster manage_security, manage_api_key ou manage_own_api_key)
  • Connectivité réseau entre les fonctions Cloud Run et votre cluster Elasticsearch

créer un bucket Google Cloud Storage ;

  1. Accédez à Google Cloud Console.
  2. Sélectionnez votre projet ou créez-en un.
  3. Dans le menu de navigation, accédez à Cloud Storage> Buckets.
  4. Cliquez sur Créer un bucket.

  5. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nommer votre bucket Saisissez un nom unique (par exemple, elastic-defend-logs).
    Type d'emplacement Choisissez en fonction de vos besoins (région, birégion ou multirégion).
    Emplacement Sélectionnez l'emplacement (par exemple, us-central1).
    Classe de stockage Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
    Access control (Contrôle des accès) Uniforme (recommandé)
    Outils de protection Facultatif : Activer la gestion des versions des objets ou la règle de conservation

  6. Cliquez sur Créer.

Collecter les identifiants Elastic Defend

Pour permettre à la fonction Cloud Run de récupérer les événements Elastic Defend, vous devez créer une clé API avec des autorisations de lecture sur les flux de données logs-endpoint.

Créer une clé API à l'aide de Kibana

  1. Connectez-vous à Kibana.
  2. Dans le menu de navigation ou le champ de recherche global, accédez à Stack Management> API Keys (Gestion de la pile > Clés API).
  3. Cliquez sur Créer une clé API.
  4. Dans le champ Nom, saisissez Google SecOps Cloud Storage Integration.
  5. Dans le champ Expiration, vous pouvez définir une date d'expiration. Par défaut, les clés API n'expirent pas.
  6. Cliquez sur Contrôler les droits d'accès à la sécurité.
  7. Dans la section Indices, cliquez sur Ajouter un droit d'accès à l'index.
  8. Configurez le privilège d'index :
    • Indices : saisissez logs-endpoint.*.
    • Droits : sélectionnez Lecture.
  9. Laissez la section Droits d'accès au cluster vide (aucun droit d'accès au cluster n'est requis).
  10. Cliquez sur Créer une clé API.

Enregistrer les identifiants de l'API

Une fois la clé API créée, une boîte de dialogue affiche vos identifiants :

  • Encodée : clé API encodée en base64 (par exemple, VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw==)

    Vous devez également enregistrer l'URL de votre point de terminaison Elasticsearch :

  • Pour Elastic Cloud : le point de terminaison s'affiche dans la section Elasticsearch de votre déploiement dans la console Cloud (par exemple, https://my-deployment.es.us-central1.gcp.cloud.es.io:443).

  • Pour Elasticsearch autogéré : utilisez le nom d'hôte ou l'adresse IP de votre cluster Elasticsearch avec le port (par exemple, https://elasticsearch.example.com:9200).

Créer une clé API à l'aide des outils pour les développeurs (méthode alternative)

Vous pouvez également créer une clé API à l'aide des outils de développement Kibana :

  1. Connectez-vous à Kibana.
  2. Accédez à Gestion > Outils de développement.

  3. Dans la console, exécutez la commande suivante :

    POST /_security/api_key
{
  "name": "Google SecOps Cloud Storage Integration",
  "role_descriptors": {
    "chronicle_reader": {
      "indices": [
        {
          "names": ["logs-endpoint.*"],
          "privileges": ["read"]
        }
      ]
    }
  },
  "metadata": {
    "application": "google-chronicle-gcs",
    "environment": "production"
  }
}

  4. La réponse contient les identifiants de votre clé API :

    {
  "id": "VuaCfGcBCdbkQm-e5aOx",
  "name": "Google SecOps Cloud Storage Integration",
  "api_key": "ui2lp2axTNmsyakw9tvNnw",
  "encoded": "VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw=="
}

  5. Copiez et enregistrez la valeur encodée. Il s'agit de la clé API encodée en base64 que vous utiliserez pour l'authentification.

Créer un compte de service

Créez un compte de service dédié pour la fonction Cloud Run.

  1. Dans la console Google Cloud, accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Nom du compte de service : saisissez elastic-defend-collector.
    • Description du compte de service : saisissez Service account for Elastic Defend log collection to GCS.
  4. Cliquez sur Créer et continuer.
  5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
    1. Cliquez sur Sélectionner un rôle, puis sélectionnez Administrateur des objets Storage (roles/storage.objectAdmin).
    2. Cliquez sur Ajouter un autre rôle, puis sélectionnez Demandeur Cloud Run (roles/run.invoker).
  6. Cliquez sur Continuer.
  7. Cliquez sur OK.

Créer un sujet Pub/Sub

Créez un sujet Pub/Sub pour déclencher la fonction Cloud Run à partir de Cloud Scheduler.

  1. Dans la console Google Cloud, accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. Fournissez les informations de configuration suivantes :
    • ID du sujet : saisissez elastic-defend-trigger.
    • Ajouter un abonnement par défaut : laissez cette option cochée.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run

Créez une fonction Cloud Run qui récupère les événements d'Elasticsearch et les écrit dans GCS.

Créer la fonction

  1. Dans la console Google Cloud, accédez à Cloud Run Functions.
  2. Cliquez sur Créer une fonction.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Environnement 2e génération
    Nom de la fonction elastic-defend-to-gcs
    Région Sélectionnez la même région que votre bucket GCS.
    Type de déclencheur Cloud Pub/Sub
    Sujet Cloud Pub/Sub Sélectionner elastic-defend-trigger
    Mémoire allouée 512 Mio (augmentation pour les grands volumes de données)
    Délai avant expiration 540 secondes
    Compte de service d'exécution Sélectionner elastic-defend-collector

  4. Cliquez sur Suivant.

Ajouter des variables d'environnement

  • Ajoutez les variables d'environnement suivantes dans la section Paramètres d'exécution, de compilation, de connexion et de sécurité :

    Variable Valeur
    GCS_BUCKET Nom de votre bucket GCS (par exemple, elastic-defend-logs)
    GCS_PREFIX Préfixe des fichiers journaux (par exemple, elastic-defend)
    STATE_KEY Nom du fichier d'état (par exemple, state.json)
    ES_HOST URL Elasticsearch (par exemple, https://my-deployment.es.us-central1.gcp.cloud.es.io:443)
    ES_API_KEY Clé API encodée à partir de l'étape de création des identifiants
    MAX_RECORDS Nombre maximal d'enregistrements par exécution (100000, par exemple)
    PAGE_SIZE Nombre d'enregistrements par requête de recherche (par exemple, 1000)
    LOOKBACK_HOURS Nombre d'heures à analyser lors de la première exécution (par exemple, 24)

Ajouter le code de la fonction

  1. Sélectionnez Python 3.11 comme environnement d'exécution.
  2. Définissez le champ Point d'entrée sur main.
  3. Dans la section Code source, sélectionnez Éditeur intégré.

  4. Remplacez le contenu de main.py par le code suivant :

    import os
import json
import datetime
import base64
import requests
from google.cloud import storage

GCS_BUCKET = os.environ["GCS_BUCKET"]
GCS_PREFIX = os.environ.get("GCS_PREFIX", "elastic-defend")
STATE_KEY = os.environ.get("STATE_KEY", "state.json")
ES_HOST = os.environ["ES_HOST"]
ES_API_KEY = os.environ["ES_API_KEY"]
MAX_RECORDS = int(os.environ.get("MAX_RECORDS", "100000"))
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "1000"))
LOOKBACK_HOURS = int(os.environ.get("LOOKBACK_HOURS", "24"))

INDEX_PATTERN = "logs-endpoint.*"
SEARCH_PATH = f"/{INDEX_PATTERN}/_search"

def _gcs_client():
    return storage.Client()

def _load_state(bucket):
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    if blob.exists():
        return json.loads(blob.download_as_text())
    return {}

def _save_state(bucket, state):
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    blob.upload_from_string(
        json.dumps(state),
        content_type="application/json",
    )

def _build_query(gte_ts, sort_after=None):
    body = {
        "size": PAGE_SIZE,
        "query": {
            "range": {
                "@timestamp": {
                    "gte": gte_ts,
                    "format": "strict_date_optional_time",
                }
            }
        },
        "sort": [
            {"@timestamp": {"order": "asc"}},
            {"_shard_doc": "asc"},
        ],
    }
    if sort_after:
        body["search_after"] = sort_after
    return body

def _search(session, body):
    url = f"{ES_HOST.rstrip('/')}{SEARCH_PATH}"
    resp = session.post(
        url,
        json=body,
        headers={
            "Authorization": f"ApiKey {ES_API_KEY}",
            "Content-Type": "application/json",
        },
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()

def _write_ndjson(bucket, records, ts_label):
    if not records:
        return
    now = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    blob_name = f"{GCS_PREFIX}/{ts_label}/{now}.ndjson"
    blob = bucket.blob(blob_name)
    ndjson = "\n".join(json.dumps(r, separators=(",", ":")) for r in records)
    blob.upload_from_string(ndjson, content_type="application/x-ndjson")
    print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{blob_name}")

def main(event, context):
    """Cloud Run function entry point triggered by Pub/Sub."""
    client = _gcs_client()
    bucket = client.bucket(GCS_BUCKET)

    state = _load_state(bucket)
    sort_after = state.get("sort_after")

    if state.get("last_timestamp"):
        gte_ts = state["last_timestamp"]
    else:
        gte_ts = (
            datetime.datetime.utcnow()
            - datetime.timedelta(hours=LOOKBACK_HOURS)
        ).strftime("%Y-%m-%dT%H:%M:%S.%fZ")

    session = requests.Session()
    total = 0
    batch = []
    last_ts = gte_ts
    ts_label = datetime.datetime.utcnow().strftime("%Y/%m/%d/%H")

    while total < MAX_RECORDS:
        body = _build_query(gte_ts, sort_after)
        result = _search(session, body)
        hits = result.get("hits", {}).get("hits", [])

        if not hits:
            break

        for hit in hits:
            doc = hit.get("_source", {})
            doc["_id"] = hit.get("_id")
            doc["_index"] = hit.get("_index")
            batch.append(doc)

            hit_ts = doc.get("@timestamp", last_ts)
            if hit_ts > last_ts:
                last_ts = hit_ts

        sort_after = hits[-1].get("sort")
        total += len(hits)

        if len(batch) >= PAGE_SIZE:
            _write_ndjson(bucket, batch, ts_label)
            batch = []

        if len(hits) < PAGE_SIZE:
            break

    if batch:
        _write_ndjson(bucket, batch, ts_label)

    new_state = {
        "last_timestamp": last_ts,
        "sort_after": sort_after,
    }
    _save_state(bucket, new_state)
    print(f"Done. Fetched {total} records. State: {json.dumps(new_state)}")
    return f"OK: {total} records"

  5. Remplacez le contenu de requirements.txt par le code suivant :

    functions-framework==3.*
google-cloud-storage==2.*
requests==2.*

  6. Cliquez sur Déployer.

  7. Attendez la fin du déploiement.

Créer une tâche Cloud Scheduler

Créez une tâche Cloud Scheduler pour déclencher la fonction Cloud Run à intervalles réguliers.

  1. Dans la console Google Cloud, accédez à Cloud Scheduler.
  2. Cliquez sur Créer une tâche.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom elastic-defend-scheduler
    Région Sélectionnez la même région que votre fonction Cloud Run.
    Fréquence */5 * * * * (toutes les cinq minutes)
    Fuseau horaire Sélectionnez votre fuseau horaire (par exemple, UTC).

  4. Cliquez sur Continuer.

  5. Dans la section Configurer l'exécution :

    • Type de cible : sélectionnez Pub/Sub.
    • Sujet Cloud Pub/Sub : sélectionnez elastic-defend-trigger.
    • Corps du message : saisissez {"run": true}.

  6. Cliquez sur Créer.

Récupérer le compte de service Google SecOps et configurer le flux

Google SecOps utilise un compte de service unique pour lire les données de votre bucket GCS. Vous devez accorder à ce compte de service l'accès à votre bucket.

Obtenir l'adresse e-mail du compte de service

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Add New Feed (Ajouter un flux).
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Elastic Defend Events).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez Elastic Defend comme Type de journal.

  7. Cliquez sur Obtenir un compte de service. Une adresse e-mail unique pour le compte de service s'affiche, par exemple :

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Copiez cette adresse e-mail pour l'utiliser à l'étape suivante.

Configurer le flux

  1. Cliquez sur Suivant.

  2. Spécifiez les valeurs des paramètres d'entrée suivants :

    • URL du bucket de stockage : saisissez l'URI du bucket GCS avec le chemin d'accès au préfixe :

      gs://elastic-defend-logs/elastic-defend/

      Remplacez elastic-defend-logs par le nom de votre bucket GCS.

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Jamais : ne supprime jamais aucun fichier après les transferts (recommandé pour les tests).
      • Supprimer les fichiers transférés : supprime les fichiers après un transfert réussi.

      • Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides après un transfert réussi.

    • Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours (180 jours par défaut).

    • Espace de noms de l'élément : espace de noms de l'élément

    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux

  3. Cliquez sur Suivant.

  4. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps a besoin du rôle Lecteur des objets Storage sur votre bucket GCS.

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (par exemple, elastic-defend-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
    • Attribuer des rôles : sélectionnez Lecteur des objets de l'espace de stockage.

  6. Cliquez sur Enregistrer.

Table de mappage UDM

Champ de journal Mappage UDM Logique
_source.agent.id, _source.agent.type, _source.agent.version, _source.host.architecture, _source.event.agent_id_status, _source.event.id, _source.user.id, _source.group.id, _source.data_stream.type, _source.agent.build.original additional.fields Fusionné avec les objets libellés contenant les valeurs des champs listés
_source.process.Ext.session_info.logon_type extensions.auth.auth_details Valeur copiée directement
_source.host.os.full hardware.cpu_platform Valeur copiée directement
_source.host.id hardware.serial_number Valeur copiée directement
_source.rule.description metadata.description Valeur copiée directement
_source.@timestamp metadata.event_timestamp Converti à l'aide du filtre de date avec les formats ISO8601, yyyy-MM-ddTHH:mm:ss.SSSSSSSSSZ, yyyy-MM-ddTHH:mm:ss.SSSSSSZ, yyyy-MM-ddTHH:mm:ss.SSSZ et yyyy-MM-ddTHH:mm:ssZ
metadata.event_type Définies en fonction des conditions index, event.action et has*
metadata.log_type Définissez-le sur "ELASTIC_DEFEND".
metadata.product_event_type Définissez la valeur sur "File Events" (Événements de fichier) si _index ~ events.file, sur "Library Events" (Événements de bibliothèque) si events.library, sur "Network Events" (Événements réseau) si events.network, sur "Process Events" (Événements de processus) si events.process, sur "Registry Events" (Événements de registre) si events.registry, sur "Security Events" (Événements de sécurité) si events.security, sur "API Events" (Événements d'API) si events.api ou sur "Alerts" (Alertes) si .alert.
_id metadata.product_log_id Valeur de _id si elle n'est pas dans ["", " ", "null", "N/A"]
_source.ecs.version metadata.product_version Valeur copiée directement
_source.network.type network.application_protocol_version Valeur copiée directement
_source.network.transport network.ip_protocol Défini sur "TCP" si la valeur correspond à (?i)tcp, sur "UDP" si elle correspond à (?i)udp, sur "ICMP" si elle correspond à (?i)icmp, ou sur "UNKNOWN_IP_PROTOCOL" dans le cas contraire.
_source.destination.as.organization.name network.organization_name Valeur copiée directement
_source.Endpoint.policy.applied.artifacts.global.identifiers observer.file.names Fusionné à partir de _source.Endpoint.policy.applied.artifacts.global.identifiers
_source.Endpoint.policy.applied.artifacts.global.version, _source.Endpoint.policy.applied.artifacts.global.snapshot observer.resource.attribute.labels Fusionné avec les objets libellés contenant les valeurs des champs listés
_source.Endpoint.policy.applied.artifacts.user.version observer.user.attribute.labels Fusionné avec l'objet d'étiquette contenant la valeur de _source.Endpoint.policy.applied.artifacts.user.version
_source.host.os.full principal.asset.hardware.cpu_platform Valeur copiée directement
_source.host.id principal.asset.hardware.serial_number Valeur copiée directement
_source.host.name principal.asset.hostname Valeur copiée directement
_source.host.ip principal.asset.ip Fusionné à partir de _source.host.ip
_source.host.os.type principal.asset.platform_software.platform Définissez sur "WINDOWS" si la valeur correspond à (?i)windows, sur "LINUX" si (?i)linux, sur "MAC" si (?i)mac, sur "IOS" si (?i)ios, ou sur "UNKNOWN_PLATFORM" dans le cas contraire.
_source.host.os.kernel principal.asset.platform_software.platform_patch_level Valeur copiée directement
_source.event.created principal.domain.creation_time Converti à l'aide du filtre de date avec les formats ISO8601, yyyy-MM-ddTHH:mm:ss.SSSSSSSSSZ, yyyy-MM-ddTHH:mm:ss.SSSSSSZ, yyyy-MM-ddTHH:mm:ss.SSSZ et yyyy-MM-ddTHH:mm:ssZ
_source.user.domain principal.domain.name Valeur copiée directement
_source.process.thread.capabilities.effective principal.file.capabilities_tags Fusionné à partir de _source.process.thread.capabilities.effective
_source.process.executable principal.file.full_path Valeur copiée directement
_source.process.hash.md5 principal.file.md5 Valeur copiée directement
_source.file.name principal.file.names Fusionné à partir de _source.file.name
_source.process.hash.sha1 principal.file.sha1 Valeur copiée directement
_source.process.hash.sha256 principal.file.sha256 Valeur copiée directement
_source.host.hostname principal.hostname Valeur copiée directement
_source.host.ip principal.ip Fusionné à partir de _source.host.ip
_source.host.mac principal.mac Fusionné à partir de _source.host.mac après avoir remplacé - par :
_source.host.os.Ext.variant principal.platform_version Valeur copiée directement
_source.source.port principal.port Converti en chaîne, puis en entier
_source.process.command_line, _source.process.name principal.process.command_line Valeur issue de _source.process.command_line si elle n'est pas vide, sinon issue de _source.process.name
_source.process.thread.capabilities.permitted principal.process.file.capabilities_tags Fusionné à partir de _source.process.thread.capabilities.permitted
_source.process.executable principal.process.file.full_path Valeur copiée directement
_source.process.hash.md5 principal.process.file.md5 Valeur copiée directement
_source.process.hash.sha1 principal.process.file.sha1 Valeur copiée directement
_source.process.hash.sha256 principal.process.file.sha256 Valeur copiée directement
_source.process.parent.executable principal.process.parent_process.file.full_path Valeur copiée directement
_source.process.pid principal.process.pid Converti en chaîne, puis copié
_source.process.Ext.api.name principal.resource.attribute.labels Fusionné avec l'objet d'étiquette contenant la valeur de _source.process.Ext.api.name
_source.event.code principal.resource.product_object_id Valeur copiée directement
_source.group.name principal.user.group_identifiers Fusionné à partir de _source.group.name
_source.user.name principal.user.userid Valeur copiée directement
_source.user.id principal.user.windows_sid Valeur de _source.user.id si elle correspond à l'expression régulière ^S-\\d-(\\\\d+-){1,14}\\\\d+$
_source.file.Ext.malware_signature.primary.signature.hash.sha256 security_result.about.file.sha256 Valeur copiée directement
_source.event.outcome security_result.action Valeur extraite de _source.event.outcome, mise en majuscules, puis définie sur ALLOW si elle est dans [SUCCESS, ALLOW], BLOCK si elle est dans [FAILURE, DENY, SKIPPED, RATE_LIMIT], UNKNOWN_ACTION si elle est UNKNOWN
_source.event.action security_result.action_details Valeur copiée directement
_source.destination.geo.region_iso_code security_result.associations Fusionné avec l'objet contenant le nom de _source.destination.geo.region_iso_code
_source.kibana.alert.rule.parameters.threat.tactic.id, _source.kibana.alert.rule.parameters.threat.tactic.name security_result.attack_details.tactics Fusionné avec l'objet contenant l'ID et le nom des champs listés
_source.kibana.alert.rule.parameters.threat.technique.id, _source.kibana.alert.rule.parameters.threat.technique.name, _source.kibana.alert.rule.parameters.threat.technique.subtechnique.id, _source.kibana.alert.rule.parameters.threat.technique.subtechnique.name security_result.attack_details.techniques Fusionné avec les objets contenant les champs id, name, subtechnique_id et subtechnique_name de la liste
_source.event.category security_result.category_details Fusionné à partir de _source.event.category
_source.kibana.alert.rule.description security_result.description Valeur copiée directement
_source.event.kind, _source.file.Ext.malware_signature.all_names, _source.file.Ext.malware_signature.identifier, _source.event.risk_score, _source.threat.tactic.reference, _source.threat.technique.reference, _source.threat.technique.subtechnique.reference security_result.detection_fields Fusionné avec les objets libellés contenant les valeurs des champs listés
_source.rule.id, _source.kibana.alert.rule.rule_id security_result.rule_id Valeur de _source.rule.id si elle n'est pas vide, sinon de _source.kibana.alert.rule.rule_id
_source.rule.name, _source.kibana.alert.rule.name security_result.rule_name Valeur de _source.rule.name si elle n'est pas vide, sinon de _source.kibana.alert.rule.name
_source.rule.ruleset security_result.rule_set Valeur copiée directement
security_result.severity Définir sur "LOW" ; si _index correspond à .alert, définir sur "HIGH" ; si _source.kibana.alert.rule.parameters.severity correspond à (?i)LOW, définir sur "LOW"
_source.message security_result.summary Valeur copiée directement
_source.file.Ext.malware_signature.primary.signature.id security_result.threat_id Valeur copiée directement
_source.file.Ext.malware_signature.primary.signature.name security_result.threat_name Valeur copiée directement
_source.source.address, _source.source.ip src.asset.ip Fusionné à partir de _source.source.address et _source.source.ip
_source.source.address, _source.source.ip src.ip Fusionné à partir de _source.source.address et _source.source.ip
_source.host.name target.asset.hostname Valeur copiée directement
_source.destination.address, _source.destination.ip target.asset.ip Fusionné à partir de _source.destination.address et _source.destination.ip
_source.file.path, _source.dll.path, _source.process.executable, _source.Target.process.executable target.file.full_path Valeur de _source.file.path si events.file, _source.dll.path si events.library, _source.process.executable si events.process ou events.api, _source.Target.process.executable si events.api
_source.dll.hash.md5, _source.process.hash.md5 target.file.md5 Valeur de _source.dll.hash.md5 si events.library, _source.process.hash.md5 si .alert
_source.dll.name, _source.process.name target.file.names Fusionné à partir de _source.dll.name si events.library, _source.process.name si .alert
_source.dll.hash.sha1, _source.process.hash.sha1 target.file.sha1 Valeur de _source.dll.hash.sha1 si events.library, _source.process.hash.sha1 si .alert
_source.dll.hash.sha256, _source.process.hash.sha256 target.file.sha256 Valeur de _source.dll.hash.sha256 si events.library, _source.process.hash.sha256 si .alert
_source.host.name target.hostname Valeur copiée directement
_source.destination.address, _source.destination.ip target.ip Fusionné à partir de _source.destination.address et _source.destination.ip
_source.destination.geo.city_name target.location.city Valeur copiée directement
_source.destination.geo.country_name target.location.country_or_region Valeur copiée directement
_source.destination.geo.continent_name target.location.name Valeur copiée directement
_source.destination.geo.location.lat target.location.region_coordinates.latitude Converti en chaîne, puis en float
_source.destination.geo.location.lon target.location.region_coordinates.longitude Converti en chaîne, puis en float
_source.destination.geo.region_name target.location.state Valeur copiée directement
_source.data_stream.namespace target.namespace Valeur copiée directement
_source.destination.port target.port Converti en chaîne, puis en entier
_source.process.command_line target.process.command_line Valeur copiée directement
_source.process.executable target.process.file.full_path Valeur copiée directement
_source.process.hash.md5 target.process.file.md5 Valeur copiée directement
_source.process.hash.sha1 target.process.file.sha1 Valeur copiée directement
_source.process.hash.sha256 target.process.file.sha256 Valeur copiée directement
_source.process.name target.process.file.names Fusionné à partir de _source.process.name
_source.registry.key target.registry.registry_key Valeur copiée directement
_source.registry.path target.registry.registry_value_data Valeur copiée directement
_source.registry.value target.registry.registry_value_name Valeur copiée directement
_source.data_stream.dataset target.resource.name Valeur copiée directement
_source.process.entity_id target.user.userid Valeur copiée directement
metadata.product_name Définissez sur "Elastic Defend".
metadata.vendor_name Défini sur "Élastique"

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.