Na parte anterior nos conhecemos com a pilha ELK, quais produtos de software ela é composta. E a primeira tarefa que um engenheiro enfrenta ao trabalhar com a pilha ELK é o envio de logs para armazenamento no Elasticsearch para análise posterior. No entanto, isso soa mais fácil do que é; o Elasticsearch armazena logs na forma de documentos com campos e valores específicos, o que significa que o engenheiro deve usar várias ferramentas para analisar a mensagem que está sendo enviada a partir dos sistemas finais. Isso pode ser feito de várias maneiras — escrevendo um programa que irá adicionar documentos ao banco de dados via API ou utilizando soluções já existentes. No âmbito deste curso, analisaremos uma solução Logstash, que é parte da pilha ELK. Vamos ver como enviar logs dos sistemas finais para o Logstash e, em seguida, configuraremos o arquivo de configuração para analisar e redirecionar para o banco de dados Elasticsearch. Para isso, usaremos logs de um firewall Check Point como sistema de entrada.
O curso não aborda a instalação da pilha ELK, pois há uma grande quantidade de artigos sobre esse tema; focaremos na parte de configuração.
Vamos elaborar um plano de ações para a configuração do Logstash:
- Verificação se o Elasticsearch aceitará logs (testando a funcionalidade e a abertura da porta).
- Analisando como podemos enviar eventos ao Logstash, escolhemos um método e o implementamos.
- Configurando o Input no arquivo de configuração do Logstash.
- Configurando o Output no arquivo de configuração do Logstash no modo de depuração, para entender como a mensagem de log se apresenta.
- Configurando o Filter.
- Configurando o Output correto no Elasticsearch.
- Iniciando o Logstash.
- Verificando os logs no Kibana.
Vamos examinar cada item em mais detalhes:
Verificação se o Elasticsearch aceitará logs
Para isso, você pode verificar com o comando curl o acesso ao Elasticsearch a partir do sistema onde o Logstash está implantado. Se você tiver autenticação configurada, também deve passar o usuário/senha pelo curl, especificando a porta 9200 se você não a alterou. Se a resposta for semelhante ao que está indicado abaixo, está tudo certo.
[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" : "Você sabe, para busca"
}
[elastic@elasticsearch ~]$
Se a resposta não chegar, pode haver vários tipos de erros: o processo do elasticsearch não está funcionando, a porta especificada está errada ou a porta está bloqueada pelo firewall no servidor onde o elasticsearch está instalado.
Vamos considerar como enviar logs para o Logstash através do firewall Check Point.
Com o servidor de gerenciamento do Check Point, é possível enviar logs para o Logstash via syslog, usando a ferramenta log_exporter; você pode encontrar mais informações sobre ela aqui. , aqui deixaremos apenas o comando que cria o fluxo:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — é o endereço do servidor onde o Logstash está sendo executado, target-port 5555 — é a porta para a qual enviaremos os logs; o envio de logs via tcp pode sobrecarregar o servidor, portanto, em alguns casos, pode ser mais adequado usar udp.
Configurando o 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 é composto por 3 partes significativas: INPUT, FILTER, OUTPUT. No INPUT especificamos de onde o sistema deverá obter os logs, em FILTER analisamos o log — configuramos como dividir a mensagem em campos e valores, em OUTPUT configuramos o fluxo de saída — para onde os logs analisados serão enviados.
Primeiro configuramos o INPUT, observamos alguns tipos que podem existir — file, tcp e exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
indica que o Logstash aceita conexões.
port => 5555
host => "10.10.1.205"
Aceitamos conexões no endereço IP 10.10.1.205 (Logstash), a porta 5555 — a porta deve ser permitida pela política do firewall.
type => "checkpoint"
Marcamos o documento, o que é muito conveniente caso você tenha várias conexões de entrada. Posteriormente, para cada conexão, é possível escrever seu próprio filtro usando a estrutura condicional if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrição das configurações:
path => "/var/log/openvas_report/*"
Especificamos o diretório cujos arquivos devem ser lidos.
type => "openvas"
Tipo de evento.
start_position => "beginning"
Ao alterar o arquivo, ele é lido completamente; se configurado como "end", o sistema espera a chegada de novas entradas no final do arquivo.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Este input executa (apenas!) um comando shell e seu resultado é encapsulado em uma mensagem de log.
command => "ls -alh"
O comando cujo resultado nos interessa.
interval => 30
Intervalo de execução do comando em segundos.
Para receber logs do firewall, configuramos um filtro tcp ou udp, dependendo de como os logs são enviados ao Logstash.
Configuramos Output no arquivo de configuração do Logstash em modo de depuração, para entender como é a mensagem de log
Após configurarmos o INPUT, precisamos entender como será a mensagem de log e quais métodos usar para configurar o filtro (parser) dos logs.
Para isso, usaremos um filtro que envia o resultado para stdout para revisar a mensagem original, o arquivo de configuração completo até este momento será assim:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec => json }
}
}
Executamos o comando para verificação:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vemos o resultado, a imagem é clicável:
Se copiar, 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="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,
Analisando as mensagens, percebemos que os logs têm o formato: campo = valor ou key = value, o que significa que o filtro chamado kv é apropriado. Para escolher corretamente o filtro para cada caso específico, seria bom se familiarizar com a documentação técnica ou perguntar a um amigo.
Configurando o Filtro
Na etapa anterior, escolhemos kv, a seguir está a configuração deste filtro:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Escolhemos o símbolo com o qual vamos dividir campo e valor — “=”. Se tivermos registros iguais nos logs, salvamos apenas uma instância na base, caso contrário, teremos um array de valores iguais, ou seja, se tivermos a mensagem “foo = some foo=some”, registramos apenas foo = some.
Configurando a saída correta no ElasticSearch
Depois que o Filtro foi configurado, podemos exportar os logs para a base 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 na base 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-" + a data e hora atual. Cada índice pode ter um conjunto específico de campos, ou é criado automaticamente ao surgir um novo campo na mensagem; as configurações de campos e seu tipo podem ser vistas nas mappings.
Se você tiver configurada a autenticação (veremos mais tarde), é essencial que as credenciais para gravação em um índice específico sejam informadas, neste exemplo, é "tssolution" com a senha "cool". É possível restringir os direitos dos usuários para gravar logs apenas em um índice específico e em nenhum outro.
Iniciamos 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 o arquivo de configuração quanto à sua correção:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Iniciamos o processo Logstash:
sudo systemctl start logstash
Verificamos se o processo foi iniciado:
sudo systemctl status logstash
Verificamos se o socket está ativo:
netstat -nat | grep 5555
Verificando os logs no Kibana.
Depois que tudo estiver em funcionamento, acesse o Kibana — Discover, e verifique se 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
Discutimos como escrever o arquivo de configuração do Logstash e, como resultado, criamos um parser para todos os campos e valores. Agora podemos trabalhar com a busca e criar gráficos para campos específicos. No próximo módulo, abordaremos a visualização no Kibana e criaremos um dashboard simples. Vale mencionar que o arquivo de configuração do Logstash precisa ser constantemente atualizado em certas situações, por exemplo, quando queremos substituir o valor de um campo de número para texto. Em artigos futuros, faremos isso regularmente.
Então fiquem de olho nas atualizações (, , , ), .
Fonte: habr.com
