straight_es.conf

# Emplacement réel : /etc/logstash/pipeline/straight_es.conf
#
# Pipeline "straight-es" — routing et enrichissement vers Elasticsearch.
# Reçoit les events depuis le pipeline skoupa via la pipe interne "to-es".
#
# Logique de routing (ordre important) :
#   1. agent.type == "winlogbeat"  → soc-winlogbeat-YYYY.MM.dd
#      Raison : Winlogbeat ne peuple JAMAIS event.module (contrairement à Filebeat
#      avec modules). Sans ce check prioritaire, les events Winlogbeat tombent dans
#      soc-%{[event][module]} qui reste un placeholder non résolu — indexation silencieuse.
#
#   2. event.module présent        → soc-{event.module}-YYYY.MM.dd
#      Couvre les modules Filebeat (auditd → soc-auditd-*, system → soc-system-*, etc.)
#
#   3. fallback                    → soc-unknown-YYYY.MM.dd
#      Diagnostic : si des events arrivent ici, vérifier la source Beats.

input {
  pipeline { address => "to-es" }
}

filter {
  # Winlogbeat ne peuple pas event.module — routing via agent.type
  if [agent][type] == "winlogbeat" {
    mutate { add_field => { "[@metadata][target_index]" => "soc-winlogbeat" } }
  } else if [event][module] {
    mutate { add_field => { "[@metadata][target_index]" => "soc-%{[event][module]}" } }
  } else {
    mutate { add_field => { "[@metadata][target_index]" => "soc-unknown" } }
  }

  # GeoIP enrichment sur les connexions réseau Sysmon (Event ID 3)
  # Peuple destination.geo.country_name et destination.geo.location (lat/lon)
  # pour les visualisations carte dans Kibana.
  if [winlog][event_data][DestinationIp] {
    geoip {
      source => "[winlog][event_data][DestinationIp]"
      target => "[destination][geo]"
    }
  }
}

output {
  elasticsearch {
    hosts                 => ["https://192.168.126.10:9200"]
    index                 => "%{[@metadata][target_index]}-%{+YYYY.MM.dd}"
    user                  => "elastic"
    password              => "<MOT_DE_PASSE_ELASTIC>"
    ssl_enabled           => true
    ssl_verification_mode => none
  }
}