2. Elastic stack: análisis de registros de seguridad. Logstash

En el anterior el artículo nos conocimos con el stack ELK, de qué productos de software está compuesto. Y la primera tarea con la que se enfrenta un ingeniero al trabajar con el stack ELK es enviar logs para su almacenamiento en Elasticsearch para su posterior análisis. Sin embargo, esto es simplemente en teoría; Elasticsearch almacena logs en forma de documentos con campos y valores específicos, lo que significa que el ingeniero debe usar diversas herramientas para analizar el mensaje que se envía desde los sistemas finales. Esto se puede hacer de varias maneras: escribir un programa que mediante API añada documentos a la base de datos o utilizar soluciones ya preparadas. En el marco de este curso, vamos a analizar la solución Logstash, que es parte del stack ELK. Veremos cómo se pueden enviar logs desde los sistemas finales a Logstash, y luego configuraremos el archivo de configuración para el análisis y redirección a la base de datos Elasticsearch. Para ello, tomaremos como sistema de entrada los logs del cortafuegos Check Point.

En el marco del curso no se considera la instalación del stack ELK, ya que existe una gran cantidad de artículos sobre este tema; vamos a considerar la parte de configuración.

Elaboraremos un plan de acción para la configuración de Logstash:

1. Verificar que Elasticsearch aceptará logs (verificar la funcionalidad y la apertura del puerto).
2. Analizaremos de qué manera podemos enviar eventos a Logstash, elegimos el método y lo implementamos.
3. Configuramos la entrada en el archivo de configuración de Logstash.
4. Configuramos la salida en el archivo de configuración de Logstash en modo depuración para entender cómo se ve el mensaje de log.
5. Configuramos el filtro.
6. Configuramos la salida correcta en Elasticsearch.
7. Iniciamos Logstash.
8. Verificamos los logs en Kibana.

Analizaremos cada punto con más detalle:

Verificar que Elasticsearch estará aceptando logs

Para ello, se puede verificar con el comando curl el acceso a Elasticsearch desde el sistema donde está desplegado Logstash. Si has configurado autenticación, también pasamos usuario/contraseña mediante curl, indicando el puerto 9200 si no lo has cambiado. Si obtienes una respuesta similar a la indicada a continuación, todo está en orden.

[elastic@elasticsearch ~]$ curl -u <> : <> -sS -XGET "<>:9200"
{
  "name" : "elastic-1",
  "cluster_name" : "project",
  "cluster_uuid" : "sQzjTTuCR8q4ZO6DrEis0A",
  "version" : {
    "number" : "7.4.1",
    "build_flavor" : "default",
    "build_type" : "rpm",
    "build_hash" : "fc0eeb6e2c25915d63d871d344e3d0b45ea0ea1e",
    "build_date" : "2019-10-22T17:16:35.176724Z",
    "build_snapshot" : false,
    "lucene_version" : "8.2.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}
[elastic@elasticsearch ~]$

Si no recibiste respuesta, podría haber varios tipos de errores: el proceso de elasticsearch no está funcionando, se indicó un puerto incorrecto o el puerto está bloqueado por el firewall en el servidor donde está elasticsearch.

Vamos a considerar cómo enviar registros a Logstash a través del firewall Check Point.

Con el servidor de gestión de Check Point, se pueden enviar registros a Logstash a través de syslog, utilizando la herramienta log_exporter, que se puede conocer mejor mediante esta el artículo, aquí dejaremos solo el comando que crea el flujo:

cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified

<> — es la dirección del servidor donde se ejecuta Logstash, target-port 5555 — es el puerto al que enviaremos los registros, enviar registros por tcp puede sobrecargar el servidor, por lo que en algunos casos es más correcto usar udp.

Configuramos INPUT en el archivo de configuración de Logstash.

Por defecto, el archivo de configuración se encuentra en el directorio /etc/logstash/conf.d/. El archivo de configuración consta de 3 partes significativas: INPUT, FILTER, OUTPUT. INPUT indicamos de dónde el sistema tomará los registros, en FILTER parseamos el registro — configuramos cómo dividir el mensaje en campos y valores, en OUTPUT configuramos el flujo de salida — a dónde se enviarán los registros analizados.

Primero configuraremos INPUT, consideraremos algunos tipos que pueden existir — file, tcp y exe.

Tcp:

input {
tcp {
    port => 5555
    host => “10.10.1.205”
    type => "checkpoint"
    mode => "server"
}
}

mode => «server»
Indica que Logstash acepta conexiones.

port => 5555
host => “10.10.1.205”
Aceptamos conexiones en la dirección IP 10.10.1.205 (Logstash), el puerto 5555 — el puerto debe estar permitido por la política del firewall.

type => «checkpoint»
Marcamos el documento, es muy conveniente en caso de que tengas varias conexiones entrantes. En el futuro, se puede escribir un filtro propio para cada conexión utilizando la estructura lógica if.

File:

input {
  file {
    path => "/var/log/openvas_report/*"
    type => "openvas"
    start_position => "beginning"
    }
}

Descripción de la configuración:
path => «/var/log/openvas_report/*»
Especificamos el directorio cuyos archivos deben ser leídos.

type => "openvas"
Tipo de evento.

start_position => "beginning"
Al cambiar el archivo, lee todo el archivo; si se establece "end", el sistema espera la aparición de nuevos registros al final del archivo.

Exec:

input {
  exec {
    command => "ls -alh"
    interval => 30
  }
}

Con esta entrada, se ejecuta (solo) un comando shell y su salida se envuelve en un mensaje de log.

command => "ls -alh"
El comando cuya salida nos interesa.

interval => 30
Intervalo de llamada del comando en segundos.

Para recibir logs del cortafuegos, definimos el filtro tcp o udp, dependiendo de cómo se envían los logs a Logstash.

Configuramos la salida en el archivo de configuración de Logstash en modo de depuración, para entender cómo se ve el mensaje de log.

Después de configurar la entrada, es necesario entender cómo será el mensaje de log, qué métodos se deben utilizar para configurar el filtro (parser) de logs.

Para ello, utilizaremos un filtro que devuelve el resultado en stdout para que podamos ver el mensaje original, el archivo de configuración completo en este momento tendrá el siguiente aspecto:

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => "10.10.1.205"
   	 }
}

output 
{
	if [type] == "checkpoint" 
       {
		stdout { codec => json }
	}
}

Ejecutamos el comando para verificar:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vemos el resultado, la imagen es clicable:

Si lo copiamos, se verá así:

action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,0x5,0xfe0a0a0a,0xc0000000}" origin="10.10.10.254" originsicname="CN=ts-spb-cpgw-01,O=cp-spb-mgmt-01.tssolution.local.kncafb" sequencenum="8" time="1576766483" version="5" context_num="1" dst="10.10.10.10" dst_machine_name="ts-spb-dc-01@tssolution.local" layer_name="TSS-Standard Security" layer_name="TSS-Standard Application" layer_uuid="dae7f01c-4c98-4c3a-a643-bfbb8fcf40f0" layer_uuid="dbee3718-cf2f-4de0-8681-529cb75be9a6" match_id="8" match_id="33554431" parent_rule="0" parent_rule="0" rule_action="Accept" rule_action="Accept" rule_name="Implicit Cleanup" rule_uid="6dc2396f-9644-4546-8f32-95d98a3344e6" product="VPN-1 & FireWall-1" proto="17" s_port="37317" service="53" service_id="domain-udp" src="10.10.1.180" ","type":"qqqqq","host":"10.10.10.250","@version":"1","port":50620}{"@timestamp":"2019-12-19T14:50:12.153Z","message":"time="1576766483" action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,

Al observar los mensajes de datos, entendemos que los registros tienen el formato: campo = valor o key = value, lo que significa que se aplica un filtro llamado kv. Para elegir correctamente el filtro para cada caso específico, sería útil revisarlo en la documentación técnica o preguntar a un amigo.

Configurando el Filtro

En la etapa anterior seleccionamos kv, a continuación se presenta la configuración de este filtro:

filter {
if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
}
}

Elegimos el símbolo por el cual vamos a dividir el campo y el valor: “=”. Si tenemos registros idénticos en el log, guardamos en la base solo una instancia; de lo contrario, acabaríamos con un arreglo de valores idénticos, es decir, si tenemos el mensaje “foo = some foo=some” solo registramos foo = some.

Configuramos la salida correcta en ElasticSearch

Una vez que se ha configurado el Filtro, podemos exportar los registros a la base. elasticsearch:

output 
{
if [type] == "checkpoint"
{
 	elasticsearch 
        {
		hosts => ["10.10.1.200:9200"]
		index => "checkpoint-%{+YYYY.MM.dd}"
    		user => "tssolution"
    		password => "cool"
  	}
}
}

Si el documento está firmado con el tipo checkpoint, guardamos el evento en la base de elasticsearch, que por defecto acepta conexiones en 10.10.1.200 en el puerto 9200. Cada documento se guarda en un índice específico, en este caso se guarda en el índice «checkpoint-» + la fecha actual. Cada índice puede tener un conjunto específico de campos, o se crea automáticamente cuando aparece un nuevo campo en el mensaje; las configuraciones de los campos y su tipo se pueden ver en mappings.

Si tienes autenticación configurada (lo veremos más adelante), deben indicarse las credenciales para escribir en un índice específico; en este ejemplo, es «tssolution» con la contraseña «cool». Se pueden restringir los derechos de los usuarios para escribir registros solo en un índice determinado y en ningún otro.

Iniciamos Logstash.

Archivo de configuración de Logstash:

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => “10.10.1.205”
   	 }
}

filter {
        if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
        }
}

output 
{
if [type] == "checkpoint"
{
 	elasticsearch 
        {
		hosts => ["10.10.1.200:9200"]
		index => "checkpoint-%{+YYYY.MM.dd}"
    		user => "tssolution"
    		password => "cool"
  	}
}
}

Verificamos el archivo de configuración por errores:
/usr/share/logstash/bin//logstash -f checkpoint.conf


Iniciamos el proceso Logstash:
sudo systemctl start logstash

Verificamos que el proceso haya iniciado:
sudo systemctl status logstash

Comprobamos si el socket está activo:
netstat -nat |grep 5555

Verificamos los logs en Kibana.

Una vez que todo esté en marcha, accedemos a Kibana — Discover, asegurándonos de que todo esté configurado correctamente y la imagen sea clicable.

¡Todos los registros están en su lugar y podemos ver todos los campos y sus valores!

Conclusión

Hemos revisado cómo escribir el archivo de configuración de Logstash, y como resultado hemos creado un parser para todos los campos y valores. Ahora podemos trabajar con la búsqueda y la creación de gráficos de campos específicos. Más adelante en el curso, examinaremos la visualización en Kibana y crearemos un sencillo dashboard. Cabe mencionar que el archivo de configuración de Logstash debe ser modificado constantemente en ciertas situaciones, por ejemplo, cuando queremos substituir el valor de un campo de un número a una palabra. En los artículos siguientes, estaremos haciendo esto regularmente.

Así que estén atentos a las actualizaciones (Telegram, Facebook, VK, TS Solution Blog), Yandex.Zen.

Fuente: habr.com