En el pasado nos conocimos pila de alces, en qué productos de software se compone. Y la primera tarea que enfrenta un ingeniero cuando trabaja con la pila ELK es enviar registros para su almacenamiento en elasticsearch para su posterior análisis. Sin embargo, esto es solo palabrería, elasticsearch almacena registros en forma de documentos con ciertos campos y valores, lo que significa que el ingeniero debe usar varias herramientas para analizar el mensaje que se envía desde los sistemas finales. Esto se puede hacer de varias maneras: escriba usted mismo un programa que agregue documentos a la base de datos utilizando la API o utilice soluciones listas para usar. En este curso consideraremos la solución. Logstash, que es parte de la pila ELK. Veremos cómo podemos enviar registros desde sistemas de punto final a Logstash y luego configuraremos un archivo de configuración para analizar y redirigir a la base de datos de Elasticsearch. Para hacer esto, tomamos registros del firewall de Check Point como sistema entrante.
El curso no cubre la instalación de la pila ELK, ya que hay una gran cantidad de artículos sobre este tema, consideraremos el componente de configuración.
Elaboremos un plan de acción para la configuración de Logstash:
- Verificar que elasticsearch acepte registros (verificar la funcionalidad y apertura del puerto).
- Consideramos cómo podemos enviar eventos a Logstash, elegir un método e implementarlo.
- Configuramos Input en el archivo de configuración de Logstash.
- Configuramos la Salida en el archivo de configuración de Logstash en modo de depuración para comprender cómo se ve el mensaje de registro.
- Configuración del filtro.
- Configurar la salida correcta en ElasticSearch.
- Se inicia Logstash.
- Comprobando los registros en Kibana.
Veamos cada punto con más detalle:
Comprobando que elasticsearch aceptará registros
Para hacer esto, puede usar el comando curl para verificar el acceso a Elasticsearch desde el sistema en el que está implementado Logstash. Si tiene configurada la autenticación, también transferimos el usuario/contraseña a través de curl, especificando el puerto 9200 si no lo ha cambiado. Si recibe una respuesta similar a la siguiente, entonces todo está en orden.
[elastic@elasticsearch ~]$ curl -u <<user_name>> : <<password>> -sS -XGET "<<ip_address_elasticsearch>>: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 se recibe la respuesta, puede haber varios tipos de errores: el proceso de elasticsearch no se está ejecutando, se especifica el puerto incorrecto o el puerto está bloqueado por un firewall en el servidor donde está instalado elasticsearch.
Veamos cómo puede enviar registros a Logstash desde un firewall de punto de control.
Desde el servidor de administración de Check Point puede enviar registros a Logstash a través de syslog usando la utilidad log_exporter, puede leer más al respecto aquí , aquí dejaremos solo el comando que crea el flujo:
cp_log_export agregar nombre check_point_syslog servidor-destino < > puerto de destino 5555 protocolo formato tcp modo de lectura genérico semi-unificado
< > - dirección del servidor en el que se ejecuta Logstash, puerto de destino 5555 - puerto al que enviaremos registros, enviar registros a través de tcp puede cargar el servidor, por lo que en algunos casos es más correcto usar udp.
Configurar INPUT en el archivo de configuración de Logstash
De forma predeterminada, el archivo de configuración se encuentra en el directorio /etc/logstash/conf.d/. El archivo de configuración consta de 3 partes significativas: ENTRADA, FILTRO, SALIDA. EN ENTRADA indicamos de dónde tomará el sistema los logs, en FILTRO analizar el registro: configure cómo dividir el mensaje en campos y valores, en SALIDA Configuramos el flujo de salida, donde se enviarán los registros analizados.
Primero, configuremos la ENTRADA, consideremos algunos de los tipos que pueden ser: archivo, tcp y exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
modo => "servidor"
Indica que Logstash está aceptando conexiones.
puerto => 5555
anfitrión => “10.10.1.205”
Aceptamos conexiones a través de la dirección IP 10.10.1.205 (Logstash), puerto 5555; el puerto debe estar permitido por la política de firewall.
tipo => "punto de control"
Marcamos el documento, muy conveniente si tienes varias conexiones entrantes. Posteriormente, para cada conexión puede escribir su propio filtro utilizando la construcción lógica if.
archivo:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descripción de la configuración:
ruta => "/var/log/openvas_report/*"
Indicamos el directorio en el que se deben leer los archivos.
tipo => "openvas"
Tipo de evento.
start_position => "comienzo"
Al cambiar un archivo, lee el archivo completo; si configura "fin", el sistema espera a que aparezcan nuevos registros al final del archivo.
Ejecutivo:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Usando esta entrada, se lanza un comando de shell (¡únicamente!) y su salida se convierte en un mensaje de registro.
comando => "ls -alh"
El comando cuya salida nos interesa.
intervalo => 30
Intervalo de invocación de comando en segundos.
Para recibir registros del firewall, registramos un filtro tcp o udp, dependiendo de cómo se envíen los registros a Logstash.
Configuramos la salida en el archivo de configuración de Logstash en modo de depuración para comprender cómo se ve el mensaje de registro.
Una vez que hayamos configurado INPUT, debemos comprender cómo se verá el mensaje de registro y qué métodos se deben utilizar para configurar el filtro de registro (analizador).
Para hacer esto, usaremos un filtro que envía el resultado a stdout para ver el mensaje original; el archivo de configuración completo en este momento se verá así:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Ejecute el comando para verificar:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vemos el resultado, se puede hacer clic en la imagen:
Si lo copias 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="[email protected]" 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 estos mensajes, entendemos que los registros se ven así: campo = valor o clave = valor, lo que significa que un filtro llamado kv es adecuado. Para elegir el filtro adecuado para cada caso concreto, sería una buena idea familiarizarse con ellos en la documentación técnica o preguntar a un amigo.
Configurar el filtro
En la última etapa seleccionamos kv, la configuración de este filtro se presenta a continuación:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Seleccionamos el símbolo por el cual dividiremos el campo y el valor - “=". Si tenemos entradas idénticas en el registro, guardamos solo una instancia en la base de datos; de lo contrario, terminará con una matriz de valores idénticos, es decir, si tenemos el mensaje “foo = some foo=some” escribimos solo foo = algunos.
Configurar la salida correcta en ElasticSearch
Una vez configurado el filtro, puede cargar registros a la base de datos. 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 de punto de control, guardamos el evento en la base de datos elasticsearch, que acepta conexiones en 10.10.1.200 en el puerto 9200 de forma predeterminada. Cada documento se guarda en un índice específico, en este caso guardamos en el índice "punto de control-" + fecha de hora actual. Cada índice puede tener un conjunto específico de campos o se crea automáticamente cuando aparece un nuevo campo en un mensaje; la configuración de los campos y su tipo se pueden ver en las asignaciones.
Si tiene configurada la autenticación (lo veremos más adelante), se deben especificar las credenciales para escribir en un índice específico, en este ejemplo es “tssolution” con la contraseña “cool”. Puede diferenciar los derechos de usuario para escribir registros solo en un índice específico y nada más.
Inicie 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 que el archivo de configuración sea correcto:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Inicie el proceso de Logstash:
sudo systemctl start logstash
Comprobamos que el proceso ha comenzado:
sudo systemctl estado logstash
Comprobemos si el enchufe está activo:
netstat-nat |grep 5555
Comprobando los registros en Kibana.
Después de que todo se esté ejecutando, vaya a Kibana - Discover, asegúrese de que todo esté configurado correctamente, ¡se puede hacer clic en la imagen!
¡Todos los registros están en su lugar y podemos ver todos los campos y sus valores!
Conclusión
Vimos cómo escribir un archivo de configuración de Logstash y, como resultado, obtuvimos un analizador de todos los campos y valores. Ahora podemos trabajar con la búsqueda y el trazado de campos específicos. A continuación, en el curso veremos la visualización en Kibana y crearemos un panel simple. Cabe mencionar que el archivo de configuración de Logstash necesita actualizarse constantemente en determinadas situaciones, por ejemplo, cuando queremos reemplazar el valor de un campo de un número a una palabra. En artículos posteriores haremos esto constantemente.
Así que estad atentos (, , , ), .
Fuente: habr.com