Como faço para alimentar o log de alerta do Elasticsearch my Snort?

Question

Eu comecei ontem com um howto ELK e colocou o ELK em funcionamento facilmente. A próxima coisa que eu queria fazer era ligar o meu log de alerta do Snort. Eu configurei o Logstash (mostrado abaixo) com um filtro e um regex Grok absolutamente desagradável para dividir todos os campos usando o grokdebug para testá-lo. Então eu liguei o snort e o log de alerta começou a ser preenchido, seguido por uma reinicialização do logstash (depois de fazer --configtest , é claro). Eu instalei o plugin ES "Head" para que eu pudesse bisbilhotar um pouco. Parece que meus alertas de snort estão sendo mapeados com meu mapeamento de syslog como foi criado no howto (foto abaixo). No ES, não consigo pesquisar usando qualquer um dos campos definidos na configuração do logstash (ids_proto, src_ip, dst_ip). Por que é isso? Preciso definir um mapeamento ou alguma outra coisa está errada aqui?

input{file{path=>"/var/log/snort/alert"
    type => "snort_tcp"  # a type to identify those logs (will need this later)
    start_position => beginning 
    ignore_older => 0      # Setting ignore_older to 0 disables file age checking so that the tutorial file is processed even though it’s older than a day. 
    sincedb_path => "/dev/null"
  }
}


filter {
 if [type] == "snort_tcp" {
      grok {
        add_tag => [ "IDS" ]
        match => [ "message", "%{MONTHNUM:month}\/%{MONTHDAY:day}-%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second}\s+\[\*\*\]\s+\[%{INT:ids_gid}\:%{INT:ids_sid}\:%{INT:ids_rev}\]\s+%{DATA:ids_proto}\s+\[\*\*\]\s+\[Classification:\s+%{DATA:ids_classification}\]\s+\[Priority:\s+%{INT:priority}\]\s+\{%{WORD:ids_proto}\}\s+%{IP:src_ip}\:%{INT:src_port}\s+\-\>\s+%{IP:dst_ip}\:%{INT:dst_port}"]
      }
    }
    geoip {
      source => "[src_ip]"
      target => "SrcGeo"
    }
    geoip {
      source => "[dst_ip]"
      target => "DstGeo"
    }
        if [priority] == "1" {
      mutate {
        add_field => { "severity" => "High" }
      }
    }
    if [priority] == "2" {
      mutate {
        add_field => { "severity" => "Medium" }
      }
    }
    if [priority] == "3" {
      mutate {
        add_field => { "severity" => "Low" }
      }
    }
    if [ids_proto] {
      if [ids_proto] =~ /^GPL/ {
        mutate {
          add_tag => [ "Snort-ET-sig" ]
          add_field => [ "ids_rule_type", "Emerging Threats" ]
        }
      }
      if [ids_proto] =~ /^ET/ {
        mutate {
          add_tag => [ "Snort-ET-sig" ]
          add_field => [ "ids_rule_type", "Emerging Threats" ]
        }
      }
      if "Snort-ET-sig" not in [tags] {
        mutate {
          add_tag => [ "Snort-sig" ]
          add_field => [ "ids_rule_type", "Snort" ]
        }
      }
    }
if "Snort-sig" in [tags] {
      if [ids_gid] == "1" {
        mutate {
          add_field => [ "Signature_Info", "http://rootedyour/.com/snortsid?sid=%{ids_sid}" ]
        }
      }
      if [ids_gid] != "1" {
        mutate {
          add_field => [ "Signature_Info", "http://rootedyour.com/snortsid?sid=%{ids_gid}-%{ids_sid}" ]
        }
      }
    }
    if "Snort-ET-sig" in [tags] {
      mutate {
        add_field => [ "Signature_Info", "http://doc.emergingthreats.net/bin/view/Main/%{ids_sid}" ]
      }
    }
  }



output 
{
  elasticsearch 
   {
    hosts => ["localhost:9200"]
    manage_template => false
    index => "snort_tcp-%{+YYYY.MM.dd}"     
  }
}

    

Answer 1

Algumas coisas aqui:

• O logstash de mapeamento padrão cria conjuntos de todos os campos de string como não analisados, o que tende a ser mais amigável para as ferramentas de visualização downstream.
• Não definir um mapeamento, como você está fazendo, usa o padrão ElasticSearch mapeamento dinâmico , que não é um bom ajuste para o Logstash.

Para testar as coisas, recomendo a seguinte seção de saída:

output 
{
 elasticsearch 
  {
    hosts => ["localhost:9200"]
    manage_template => true
    index => "logstash-%{+YYYY.MM.dd}"     
  }
}

Quando definido dessa forma, os índices do logstash obterão os mapeamentos padrão do logstash, que podem se aproximar do que você espera. Se for esse o caso, você provavelmente terá que definir um arquivo de mapeamento.

output 
{
 elasticsearch 
  {
    hosts => ["localhost:9200"]
    manage_template => true
    index => "snort_tcp-%{+YYYY.MM.dd}"
    template => "/etc/logstash/template.json"
    template_name => "snort_tcp"
  }
}