Configuration ELK Stack

input {
  beats {
    port => 5044
    host => "10.10.10.20"
    ssl_enabled => false
    # Seul Cowrie (10.10.20.10) autorisé par pfSense
    # Règle DMZ-001 : 10.10.20.10 → 10.10.10.20:5044
  }
}

filter {
  if [fields][log_type] == "cowrie" or [log][file][path] =~ "cowrie" {
    json {
      source => "message"
      target => "cowrie"
      tag_on_failure => ["_cowrie_json_parse_failure"]
    }

    # Extraction des champs principaux
    if [cowrie][src_ip] {
      mutate {
        add_field => {
          "attacker_ip" => "%{[cowrie][src_ip]}"
          "event_type" => "%{[cowrie][eventid]}"
          "session_id" => "%{[cowrie][session]}"
        }
      }
    }

    # Enrichissement GeoIP
    if [attacker_ip] {
      geoip {
        source => "attacker_ip"
        target => "attacker_geo"
        database => "/usr/share/GeoIP/GeoLite2-City.mmdb"
      }
    }

    # Classification des événements
    if [event_type] == "cowrie.login.success" {
      mutate { add_field => { "alert_level" => "HIGH" } }
    } else if [event_type] == "cowrie.login.failed" {
      mutate { add_field => { "alert_level" => "MEDIUM" } }
    } else if [event_type] == "cowrie.command.input" {
      mutate { add_field => { "alert_level" => "HIGH" } }
    } else if [event_type] == "cowrie.session.file_download" {
      mutate { add_field => { "alert_level" => "CRITICAL" } }
    }

    date {
      match => ["[cowrie][timestamp]", "ISO8601"]
      target => "@timestamp"
      timezone => "Europe/Paris"
    }

    mutate {
      remove_field => ["message", "host", "agent", "ecs"]
    }
  }
}

output {
  if [event_type] {
    elasticsearch {
      hosts => ["https://10.10.10.20:9200"]
      index => "cowrie-logs-%{+YYYY.MM.dd}"
      user => "elastic"
      password => "Elk$oc2026!Secure"
      ssl_enabled => true
      ssl_certificate_verification => false
      ilm_enabled => true
      ilm_rollover_alias => "cowrie-logs"
      ilm_policy => "cowrie-retention-90d"
    }
  }
}