ELK 6.6 Configurando ElastiFlow (Netflow)

Para añadir funcionalidad a nuestra reciente instalación de ELK 6.6, para ello vamos a hacer uso del plugin ElastiFlow que el propio Elastic está implantando también como propio en la instalación nativa. Puedes obtener toda la información desde su repositorio de Github.

Como hemos comentado, partimos de la instalación anterior del STACK ELK (ElasticSearch, Logstash y Kibana) en su versión 6.6, el plugin es compatible con esta versión, veamos los pasos a seguir para ponerlo en marcha.

En primer lugar echamos un vistazo a los requisitos de máquina en función de la cantidad de flows/s para dimensionar bien nuestra instalación:

también hemos de subir la memoria JVM:

vi /etc/elasticsearch/jvm.options

Cambiamos los parametros para aumentarla a 4 GB:

-Xms4g
-Xmx4g

Comprobamos que tenemos instalados y actualizados los siguientes plugins:

./usr/share/logstash/bin/logstash-plugin install logstash-codec-sflow
./usr/share/logstash/bin/logstash-plugin update logstash-codec-netfloww
./usr/share/logstash/bin/logstash-plugin update logstash-input-udp
./usr/share/logstash/bin/logstash-plugin update logstash-filter-dns

Comenzamos ya con la instalación del plugin clonando el repositorio de git en nuestro Ubuntu 18:

git clone https://github.com/robcowart/elastiflow.git

cd elasticflow

sudo cp -a logstash/elastiflow/. /etc/logstash/elastiflow/
sudo cp -a logstash.service.d/. /etc/systemd/system/logstash.service.d/

sudo systemctl daemon-reload
sudo systemctl restart logstash

Añadir el pipeline de netflow (a partir de la v6 ELK soporta multiples pipelines)

vi /etc/logstash/pipelines.yml

Añadimos la pipeline haciendo referencia a los ficheros de configuración que hemos copiado anteriormente:


- pipeline.id: elastiflow

  path.config: "/etc/logstash/elastiflow/conf.d/*.conf"

Para la definición de las variables de entorno en mi caso prefiero hacerlo manualmente:

export ELASTIFLOW_KEEP_ORIG_DATA=true
export ELASTIFLOW_TEMPLATE_PATH=/etc/logstash/elastiflow/templates
export ELASTIFLOW_ES_SSL_VERIFY=false
export ELASTIFLOW_NAMESERVER=127.0.0.1
export ELASTIFLOW_NETFLOW_UDP_RCV_BUFF=33554432
export ELASTIFLOW_GEOIP_DB_PATH=/etc/logstash/elastiflow/geoipdbs
export ELASTIFLOW_NETFLOW_UDP_QUEUE_SIZE=4096
export ELASTIFLOW_DNS_HIT_CACHE_TTL=900
export ELASTIFLOW_DICT_PATH=/etc/logstash/elastiflow/dictionaries
export ELASTIFLOW_NETFLOW_IPV4_HOST=0.0.0.0
export ELASTIFLOW_DNS_HIT_CACHE_SIZE=25000
export ELASTIFLOW_RESOLVE_IP2HOST=false
export ELASTIFLOW_DNS_FAILED_CACHE_SIZE=75000
export ELASTIFLOW_DEFAULT_APPID_SRCTYPE=__UNKNOW
export ELASTIFLOW_ES_SSL_ENABLE=false
export ELASTIFLOW_ASN_LOOKUP=true
export ELASTIFLOW_NETFLOW_UDP_WORKERS=4
export ELASTIFLOW_SFLOW_IPV4_HOST=0.0.0.0
export ELASTIFLOW_NETFLOW_IPV4_PORT=9995
export ELASTIFLOW_DEFINITION_PATH=/etc/logstash/elastiflow/definitions
export ELASTIFLOW_GEOIP_CACHE_SIZE=8192
export ELASTIFLOW_GEOIP_LOOKUP=true
export ELASTIFLOW_DNS_FAILED_CACHE_TTL=3600

Para probar que esto funcione correctamente reiniciamos logstash y comprobamos el log para ver si arranca correctamente sin errores:

systemctl restart logstash.service

tail -f /var/log/logstash/logstash-plain.log


[2019-02-19T14:44:06,287][INFO ][logstash.inputs.udp      ] UDP listener started {:address=>"0.0.0.0:9995", :receive_buffer_bytes=>"212992", :queue_size=>"4096"}
[2019-02-19T14:44:06,307][WARN ][logstash.inputs.udp      ] Unable to set receive_buffer_bytes to desired size. Requested 33554432 but obtained 212992 bytes.
[2019-02-19T14:44:06,307][WARN ][logstash.inputs.udp      ] Unable to set receive_buffer_bytes to desired size. Requested 33554432 but obtained 212992 bytes.
[2019-02-19T14:44:06,336][INFO ][logstash.inputs.udp      ] UDP listener started {:address=>"0.0.0.0:4739", :receive_buffer_bytes=>"212992", :queue_size=>"4096"}
[2019-02-19T14:44:06,337][INFO ][logstash.inputs.udp      ] UDP listener started {:address=>"0.0.0.0:6343", :receive_buffer_bytes=>"212992", :queue_size=>"4096"}
[2019-02-19T14:44:06,340][INFO ][logstash.agent           ] Pipelines running {:count=>2, :running_pipelines=>[:elastiflow, :main], :non_running_pipelines=>[]}
[2019-02-19T14:44:09,251][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Ahora tenemos que enviar Netflow a nuestro ELK, pondré a continuación la configuración básica de un switch CISCO:

Creamos el Record con los campos que vamos a enviar:

flow record flowrecord1
 description IPv4Flow
 match datalink vlan input
 match datalink mac source address input
 match datalink mac destination address input
 match ipv4 tos
 match ipv4 ttl
 match ipv4 protocol
 match ipv4 source address
 match ipv4 destination address
 match transport source-port
 match transport destination-port
 match interface input
 match flow direction
 collect transport tcp flags
 collect interface output
 collect counter bytes long
 collect counter packets long
 collect timestamp absolute first
 collect timestamp absolute last
 collect counter bytes layer2 long

Creamos el exporter que dice donde lo vamos a enviar:

flow exporter ELK71
 description netflow-elk
 destination 10.x.x.x
 source Vlan10
 transport udp 9995

Y el monitor que aplicaremos a la interfaz en sentido entrada, aquí podemos indicar varios “exporter” para enviar a varios sitios al mismo tiempo con el mismo monitor:

flow monitor monitor1
 description monitor prime elk
 exporter ELK71
 cache timeout active 60
 record flowrecord1

Luego aplicamos el monitor a una interfaz:

interface GigabitEthernet1/0/1
ip flow monitor FlowMonitor input

Y con esto debemos ver como llegan los flujos al ELK desde este equipo que acabamos de configurar.

Saludos.