No passado nós conhecemos Pilha ELK, em quais produtos de software ele consiste. E a primeira tarefa que um engenheiro enfrenta ao trabalhar com a pilha ELK é enviar logs para armazenamento no elasticsearch para análise posterior. No entanto, isso é apenas da boca para fora, o elasticsearch armazena logs na forma de documentos com determinados campos e valores, o que significa que o engenheiro deve usar várias ferramentas para analisar a mensagem que é enviada dos sistemas finais. Isso pode ser feito de várias maneiras - escreva você mesmo um programa que adicionará documentos ao banco de dados usando a API ou use soluções prontas. Neste curso consideraremos a solução Logstash, que faz parte da pilha ELK. Veremos como podemos enviar logs de sistemas endpoint para o Logstash e, em seguida, configuraremos um arquivo de configuração para analisar e redirecionar para o banco de dados Elasticsearch. Para fazer isso, consideramos os logs do firewall Check Point como o sistema de entrada.
O curso não aborda a instalação da pilha ELK, pois há um grande número de artigos sobre esse assunto; consideraremos o componente de configuração.
Vamos traçar um plano de ação para configuração do Logstash:
- Verificando se o elasticsearch aceitará logs (verificando a funcionalidade e abertura da porta).
- Consideramos como podemos enviar eventos para o Logstash, escolher um método e implementá-lo.
- Configuramos o Input no arquivo de configuração do Logstash.
- Configuramos a saída no arquivo de configuração Logstash no modo de depuração para entender a aparência da mensagem de log.
- Configurando Filtro.
- Configurando a saída correta no ElasticSearch.
- O Logstash é iniciado.
- Verificando os logs no Kibana.
Vejamos cada ponto com mais detalhes:
Verificando se o elasticsearch aceitará logs
Para fazer isso, você pode usar o comando curl para verificar o acesso ao Elasticsearch do sistema no qual o Logstash está implantado. Se você tiver a autenticação configurada, também transferimos o usuário/senha via curl, especificando a porta 9200 caso você não a tenha alterado. Se você receber uma resposta semelhante à abaixo, então está tudo em ordem.
[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 a resposta não for recebida, pode haver vários tipos de erros: o processo do elasticsearch não está em execução, a porta errada foi especificada ou a porta está bloqueada por um firewall no servidor onde o elasticsearch está instalado.
Vejamos como você pode enviar logs para o Logstash a partir de um firewall de ponto de verificação
Do servidor de gerenciamento Check Point você pode enviar logs para o Logstash via syslog usando o utilitário log_exporter, você pode ler mais sobre isso aqui , aqui deixaremos apenas o comando que cria o stream:
cp_log_export adicionar nome check_point_syslog servidor-alvo < > porta de destino 5555 protocolo tcp formato genérico modo de leitura semi-unificado
< > - endereço do servidor no qual o Logstash roda, target-port 5555 - porta para a qual enviaremos logs, enviar logs via tcp pode carregar o servidor, então em alguns casos é mais correto usar udp.
Configurando INPUT no arquivo de configuração do Logstash
Por padrão, o arquivo de configuração está localizado no diretório /etc/logstash/conf.d/. O arquivo de configuração consiste em 3 partes significativas: INPUT, FILTER, OUTPUT. EM INPUT
indicamos de onde o sistema obterá os logs, em FILTRO analisar o log - configure como dividir a mensagem em campos e valores, em SAÍDA configuramos o fluxo de saída - para onde os logs analisados serão enviados.
Primeiro, vamos configurar o INPUT, considere alguns dos tipos que podem ser - arquivo, tcp e exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
modo => "servidor"
Indica que o Logstash está aceitando conexões.
porta => 5555
anfitrião => “10.10.1.205”
Aceitamos conexões via endereço IP 10.10.1.205 (Logstash), porta 5555 – a porta deve ser permitida pela política de firewall.
tipo => "ponto de verificação"
Marcamos o documento, muito conveniente se você tiver várias conexões de entrada. Posteriormente, para cada conexão você pode escrever seu próprio filtro usando a construção lógica if.
arquivo:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrição das configurações:
caminho => "/var/log/openvas_report/*"
Indicamos o diretório no qual os arquivos precisam ser lidos.
digite => "openvas"
Tipo de evento.
posição_inicial => "início"
Ao alterar um arquivo, ele lê o arquivo inteiro; se definir “fim”, o sistema espera que novos registros apareçam no final do arquivo.
Execução:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Usando esta entrada, um (apenas!) comando shell é lançado e sua saída é transformada em uma mensagem de log.
comando => "ls -alh"
O comando cuja saída estamos interessados.
intervalo => 30
Intervalo de chamada de comando em segundos.
Para receber logs do firewall, cadastramos um filtro tcp ou udp, dependendo de como os logs são enviados ao Logstash.
Configuramos a saída no arquivo de configuração Logstash no modo de depuração para entender a aparência da mensagem de log
Depois de configurarmos o INPUT, precisamos entender como será a mensagem de log e quais métodos precisam ser usados para configurar o filtro de log (analisador).
Para fazer isso, usaremos um filtro que envia o resultado para stdout para visualizar a mensagem original; o arquivo de configuração completo no momento ficará assim:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Execute o comando para verificar:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vemos o resultado, a imagem é clicável:
Se você copiá-lo, ficará assim:
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,
Olhando para essas mensagens, entendemos que os logs se parecem com: campo = valor ou chave = valor, o que significa que um filtro chamado kv é adequado. Para escolher o filtro adequado para cada caso específico, convém familiarizar-se com eles na documentação técnica ou perguntar a um amigo.
Configurando Filtro
Na última etapa selecionamos kv, a configuração deste filtro é apresentada a seguir:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Selecionamos o símbolo pelo qual dividiremos o campo e o valor - “=”. Se tivermos entradas idênticas no log, salvamos apenas uma instância no banco de dados, caso contrário você acabará com um array de valores idênticos, ou seja, se tivermos a mensagem “foo = some foo=some” escrevemos apenas foo = alguns.
Configurando a saída correta no ElasticSearch
Depois que o Filtro estiver configurado, você poderá fazer upload de logs para o banco de dados elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Se o documento for assinado com o tipo checkpoint, salvamos o evento no banco de dados elasticsearch, que aceita conexões em 10.10.1.200 na porta 9200 por padrão. Cada documento é salvo em um índice específico, neste caso salvamos no índice “checkpoint-” + hora atual. Cada índice pode ter um conjunto específico de campos ou é criado automaticamente quando um novo campo aparece em uma mensagem; as configurações dos campos e seus tipos podem ser visualizados em mapeamentos.
Se você tiver a autenticação configurada (veremos isso mais tarde), deverão ser especificadas as credenciais para gravação em um índice específico, neste exemplo é “tssolution” com a senha “cool”. Você pode diferenciar os direitos do usuário para gravar logs apenas em um índice específico e nada mais.
Inicie o Logstash.
Arquivo de configuração do 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 a exatidão do arquivo de configuração:
/usr/share/logstash/bin//logstash -f ponto de verificação.conf
Inicie o processo Logstash:
sudo systemctl iniciar logstash
Verificamos se o processo foi iniciado:
logstash de status do sudo systemctl
Vamos verificar se o soquete está ativo:
netstat -nat |grep 5555
Verificando os logs no Kibana.
Depois que tudo estiver rodando, vá em Kibana - Discover, certifique-se de que tudo está configurado corretamente, a imagem é clicável!
Todos os logs estão no lugar e podemos ver todos os campos e seus valores!
Conclusão
Vimos como escrever um arquivo de configuração do Logstash e, como resultado, obtivemos um analisador de todos os campos e valores. Agora podemos trabalhar com busca e plotagem de campos específicos. A seguir no curso, veremos a visualização no Kibana e criaremos um painel simples. Vale ressaltar que o arquivo de configuração do Logstash precisa ser atualizado constantemente em determinadas situações, por exemplo, quando queremos substituir o valor de um campo de um número por uma palavra. Nos artigos subsequentes faremos isso constantemente.
Então fique ligado (, , , ), .
Fonte: habr.com