No último coñecémonos pila ELK, de que produtos de software consiste. E a primeira tarefa á que se enfronta un enxeñeiro cando traballa coa pila ELK é enviar rexistros para almacenalos en elasticsearch para a súa posterior análise. Non obstante, este é só un servizo de boca, elasticsearch almacena rexistros en forma de documentos con certos campos e valores, o que significa que o enxeñeiro debe usar varias ferramentas para analizar a mensaxe que se envía desde os sistemas finais. Isto pódese facer de varias maneiras: escriba vostede mesmo un programa que engadirá documentos á base de datos mediante a API ou use solucións preparadas. Neste curso consideraremos a solución logstash, que forma parte da pila ELK. Veremos como podemos enviar rexistros desde os sistemas de punto final a Logstash e, a continuación, configuraremos un ficheiro de configuración para analizar e redirixir á base de datos Elasticsearch. Para iso, tomamos rexistros do firewall de Check Point como sistema de entrada.
O curso non abarca a instalación de ELK stack, xa que hai un gran número de artigos sobre este tema; consideraremos o compoñente de configuración.
Elaboremos un plan de acción para a configuración de Logstash:
- Comprobando que elasticsearch acepta rexistros (comprobando a funcionalidade e a apertura do porto).
- Consideramos como podemos enviar eventos a Logstash, escoller un método e implementalo.
- Configuramos a entrada no ficheiro de configuración de Logstash.
- Configuramos a saída no ficheiro de configuración de Logstash en modo de depuración para comprender como é a mensaxe de rexistro.
- Configurando o filtro.
- Configurando a saída correcta en ElasticSearch.
- Lanzamento de Logstash.
- Comprobando os rexistros en Kibana.
Vexamos cada punto con máis detalle:
Comprobando que elasticsearch aceptará rexistros
Para iso, pode usar o comando curl para comprobar o acceso a Elasticsearch desde o sistema no que está implantado Logstash. Se tes a autenticación configurada, tamén transferimos o usuario/contrasinal mediante curl, especificando o porto 9200 se non o cambiou. Se recibes unha resposta semellante á seguinte, todo está en orde.
[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 ~]$
Se non se recibe a resposta, pode haber varios tipos de erros: o proceso de elasticsearch non se está executando, especifícase o porto incorrecto ou o porto está bloqueado por un firewall no servidor onde está instalado elasticsearch.
Vexamos como pode enviar rexistros a Logstash desde un firewall de punto de verificación
Desde o servidor de xestión de Check Point pode enviar rexistros a Logstash mediante syslog usando a utilidade log_exporter, pode ler máis sobre iso aquí , aquí deixaremos só o comando que crea o fluxo:
cp_log_export engadir o nome check_point_syslog target-server < > protocolo target-port 5555 formato tcp modo de lectura xenérico semi-unificado
< > - enderezo do servidor no que se executa Logstash, destino-porto 5555 - porto ao que enviaremos rexistros, o envío de rexistros a través de tcp pode cargar o servidor, polo que nalgúns casos é máis correcto usar udp.
Configurando INPUT no ficheiro de configuración de Logstash
Por defecto, o ficheiro de configuración está situado no directorio /etc/logstash/conf.d/. O ficheiro de configuración consta de 3 partes significativas: INPUT, FILTER, OUTPUT. EN ENTRADA indicamos de onde levará o sistema os rexistros, en FILTER analizar o rexistro: configura como dividir a mensaxe en campos e valores SAÍDA configuramos o fluxo de saída, onde se enviarán os rexistros analizados.
Primeiro, configuremos INPUT, consideremos algúns dos tipos que poden ser: ficheiro, tcp e exe.
Tcp:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
modo => "servidor"
Indica que Logstash está aceptando conexións.
porto => 5555
host => "10.10.1.205"
Aceptamos conexións a través do enderezo IP 10.10.1.205 (Logstash), porto 5555 - o porto debe estar permitido pola política de firewall.
tipo => "punto de control"
Marcamos o documento, moi cómodo se tes varias conexións entrantes. Posteriormente, para cada conexión pode escribir o seu propio filtro usando a construción lóxica if.
Ficheiro:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrición da configuración:
ruta => "/var/log/openvas_report/*"
Indicamos o directorio no que hai que ler os ficheiros.
type => "openvas"
Tipo de evento.
start_position => "comezo"
Ao cambiar un ficheiro, le todo o ficheiro; se estableces "fin", o sistema agarda a que aparezan novos rexistros ao final do ficheiro.
Executivo:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Usando esta entrada, lánzase un comando de shell (só!) e a súa saída convértese nunha mensaxe de rexistro.
comando => "ls -alh"
O comando cuxa saída nos interesa.
intervalo => 30
Intervalo de invocación de comandos en segundos.
Para recibir rexistros do firewall, rexistramos un filtro tcp ou pud, dependendo de como se envíen os rexistros a Logstash.
Configuramos a saída no ficheiro de configuración de Logstash en modo de depuración para comprender como é a mensaxe de rexistro
Despois de ter configurado INPUT, necesitamos entender como será a mensaxe de rexistro e que métodos hai que usar para configurar o filtro de rexistro (analizador).
Para iso, utilizaremos un filtro que envía o resultado a stdout para ver a mensaxe orixinal; o ficheiro de configuración completo neste momento terá o seguinte aspecto:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Executa o comando para comprobar:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vemos o resultado, a imaxe é clicable:
Se o copias quedará 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,
Mirando estas mensaxes, entendemos que os rexistros semellan: campo = valor ou clave = valor, o que significa que é adecuado un filtro chamado kv. Para escoller o filtro axeitado para cada caso concreto, sería unha boa idea familiarizarse con eles na documentación técnica ou preguntar a un amigo.
Configurando o filtro
Na última etapa que seleccionamos kv, a configuración deste filtro preséntase a continuación:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Seleccionamos o símbolo polo que dividiremos o campo e o valor - "=". Se temos entradas idénticas no rexistro, só gardamos unha instancia na base de datos, se non, acabará cunha matriz de valores idénticos, é dicir, se temos a mensaxe "foo = some foo=some" escribimos só foo = algúns.
Configurando a saída correcta en ElasticSearch
Unha vez configurado o filtro, pode cargar rexistros na base de datos busca elástica:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Se o documento está asinado co tipo de punto de control, gardamos o evento na base de datos elasticsearch, que acepta conexións o 10.10.1.200 no porto 9200 por defecto. Cada documento gárdase nun índice específico, neste caso gardamos no índice “checkpoint-” + data de hora actual. Cada índice pode ter un conxunto específico de campos ou créase automaticamente cando aparece un campo novo nunha mensaxe; a configuración dos campos e o seu tipo pódense ver nas asignacións.
Se tes a autenticación configurada (verémolo máis adiante), hai que especificar as credenciais para escribir nun índice específico, neste exemplo é “tssolution” co contrasinal “cool”. Podes diferenciar os dereitos dos usuarios para escribir rexistros só nun índice específico e non máis.
Inicie Logstash.
Ficheiro 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"
}
}
}
Comprobamos a corrección do ficheiro de configuración:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Inicia o proceso de Logstash:
sudo systemctl comeza o logstash
Comprobamos que o proceso comezou:
sudo systemctl status logstash
Comprobamos se o socket está activado:
netstat -nat |grep 5555
Comprobando os rexistros en Kibana.
Despois de que todo estea funcionando, vai a Kibana - Descubrir, asegúrate de que todo estea configurado correctamente, pódese facer clic na imaxe.
Todos os rexistros están no seu lugar e podemos ver todos os campos e os seus valores.
Conclusión
Observamos como escribir un ficheiro de configuración de Logstash e, como resultado, obtivemos un analizador de todos os campos e valores. Agora podemos traballar coa procura e o trazado de campos específicos. A continuación no curso veremos a visualización en Kibana e crearemos un panel sinxelo. Cabe mencionar que o ficheiro de configuración de Logstash debe actualizarse constantemente en determinadas situacións, por exemplo, cando queremos substituír o valor dun campo dun número a unha palabra. En artigos posteriores farémolo constantemente.
Así que estade atentos (, , , ), .
Fonte: www.habr.com