Coletamos logs do firewall Mikrotik em um banco de dados

Boa tarde

Quero dizer com que facilidade e natural você pode configurar um servidor de coleta de metadados de tráfego de rede para roteadores Mikrotik.

Objetivo: O objetivo será armazenar os logs de firewall "mastigados" em um banco de dados para análise posterior.

Significa: Qualquer distribuição Linux nova com o rsyslogd v8 e superior é adequada para implementação, talvez a sintaxe proposta também funcione na v7. Também precisamos de um DBMS, eu escolhi o mariadb. O crescimento do banco de dados varia de acordo com o número de regras registradas, porque o tamanho da unidade é a seu critério, no meu caso, são registradas 30 a 40 regras, o que equivale a 1200 mil linhas por dia. Durante o mês de uso do banco de dados, incluindo índices, ele cresceu para 3,8 GB.

Mecânica: O roteador envia um log para o servidor remoto via UDP. Usando expressões regulares, o servidor rsyslog limpa seqüências de informações desnecessárias, gera uma inserção SQL e a envia ao DBMS. O DBMS, usando um gatilho antes da inserção, executa limpeza e separação adicionais de campos que não puderam ser analisados ​​no rsyslog.

Configurar o RSYSLOG


Editando o arquivo /etc/rsyslog.conf
Adicione as seguintes linhas lá:

module(load="ommysql") module(load="imudp") input(type="imudp" port="514") 

Assim, carregamos os módulos necessários e abrimos a porta 514 UDP.

A linha de log do Mikrotik fica assim:

 20180927155341 BLOCKSMKNETS forward: in:ether6 - LocalTORF out:VLAN55 - RT_INET, src-mac 00:15:17:31:b8:d7, proto TCP (SYN), 192.168.0.234:2457->192.168.6.14:65535, len 60 

Como você pode ver, muitos extras para armazenamento no banco de dados e uma seleção clara serão difíceis.
Em teoria, preciso adicionar esses dados:

 20180927155341 ether6 VLAN5 192.168.0.234 2457 192.168.6.14 65535 00:15:17:31:b8:d7 TCP SYN forward BLOCKSMKNETS 60 

Não consegui obter essa linha usando apenas um rsyslog. Os usuários regulares do Rsyslog usam POSIX ERE / BRE, portanto, não há como aplicar recursos como lookahead ou lookbehind.

Existe uma ferramenta que permite depurar usuários regulares, tente, talvez você possa separar a porta do endereço, bem como o nome da interface entre in: e out:. Lembre-se de que alguns protocolos de esporte e de dados estão ausentes.

Em geral, minha saída foi a seguinte:

 20180927155341 in:ether6 out:VLAN5 192.168.0.234:2457 192.168.6.14:65535 00:15:17:31:b8:d7 TCP (SYN) forward BLOCKSMKNETS 60 

documentação sobre como preparar os regulares do rsyslog.

No formato final, o arquivo de configuração para receber o log do Mikrotik /etc/rsyslog.d/20-remote.conf terá a seguinte aparência:

 $template tpl_traflog,"insert into traflog.traffic (datetime, inif, outif, src, dst, smac, proto, flags, chain, logpref, len) values ('%timereported:::date-mysql%', '%msg:R,ERE,0,DFLT,0:in:[a-zA-Z]+[0-9]+|in:<[a-zA-Z]+-[a-zA-Z]+>--end%', '%msg:R,ERE,0,BLANK,0:out:[a-zA-Z]+[0-9]+|out:<[a-zA-Z]+-[a-zA-Z]+>--end%', '%msg:R,ERE,0,DFLT,0:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end%', '%msg:R,ERE,0,DFLT,1:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end%', '%msg:R,ERE,0,BLANK:([0-f]+:){5}[0-f]+--end%', '%msg:R,ERE,0,BLANK:\b[AX]{3,4}\b--end%', '%msg:R,ERE,0,BLANK:\([AZ]+\)|\(([AZ]+\,){1,3}[AZ]+\)--end%', '%msg:R,ERE,0,DFLT:[ax]+--end%', '%msg:F,32:2%', '%msg:R,ERE,0,DFLT:[0-9]+$--end%' )",SQL if ($fromhost-ip == '192.168.0.230') and ($syslogtag contains "firewall") then {action(type="ommysql" server="localhost" serverport="3306" db="traflog" uid="rsyslogger" pwd="rsyslogger" template="tpl_traflog") stop} 

Na primeira linha, a descrição do modelo (modelo) é uma linha de código SQL para transferi-lo para o DBMS.
A segunda linha é a condição em que a ação ocorrerá, ou seja, o registro no DBMS.
A condição se parece com a seguinte: se a origem do log = 192.168.0.230 ( if ($fromhost-ip == '192.168.0.230') ) E se a linha msg contiver “firewall” (e ($ syslogtag contém “firewall”)), use o módulo ommysql com parâmetros de conexão ( then {action(type="ommysql" server="localhost" serverport="3306" db="traflog" uid="rsyslogger" pwd="..." ) chamamos o modelo tpl_traflog ( template="tpl_traflog") ) e, depois disso, paramos o processamento adicional da linha ( stop} ).

É possível que, no seu caso, algo dê errado, seja devido aos nomes das interfaces ou prefixos de log, talvez algo mais. Para depuração, faremos o seguinte, comentaremos na segunda linha, adicionaremos um novo modelo e duas novas condições:

 $template tpl_traflog_test,"%timereported:::date-mysql% %msg:R,ERE,0,DFLT,0:in:[a-zA-Z]+[0-9]+|in:<[a-zA-Z]+-[a-zA-Z]+>--end% %msg:R,ERE,0,BLANK,0:out:[a-zA-Z]+[0-9]+|out:<[a-zA-Z]+-[a-zA-Z]+>--end% %msg:R,ERE,0,DFLT,0:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end% %msg:R,ERE,0,DFLT,1:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end% %msg:R,ERE,0,BLANK:([0-f]+:){5}[0-f]+--end% %msg:R,ERE,0,BLANK:\b[AX]{3,4}\b--end% %msg:R,ERE,0,BLANK:\([AZ]+\)|\(([AZ]+\,){1,3}[AZ]+\)--end% %msg:R,ERE,0,DFLT:[ax]+--end% %msg:F,32:2% %msg:R,ERE,0,DFLT:[0-9]+$--end%\n" if ($fromhost-ip == '192.168.0.230') then {action(type="omfile" file="/var/log/remote/192.168.0.230.log" )} if ($fromhost-ip == '192.168.0.230') then {action(type="omfile" file="/var/log/remote/192.168.0.230.log" template="tpl_traflog_test" ) stop} 

Reinicie o logger.

O modelo tpl_traflog_test é semelhante ao tpl_traflog, mas sem o SQL INSERT.

A primeira condição adiciona a linha não processada% msg% ao arquivo /var/log/remote/192.168.0.230.log, porque o modelo não está especificado.

A segunda condição adiciona a linha processada ao mesmo arquivo.
Portanto, será mais conveniente comparar.
Em seguida, prepare o banco de dados.

Preparamos um banco de dados


Abaixaremos a configuração do DBMS, tudo é padrão aqui.

Iniciamos o console do mysql e executamos o seguinte código:

 --   create database traflog character set utf8 collate utf8_bin; use traflog; --  create table traffic (id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, datetime DATETIME, inif VARCHAR(20), outif VARCHAR(20), src VARCHAR(21), sport INT(5), dst VARCHAR(21), dport INT(5), smac VARCHAR(17), proto VARCHAR(4), flags VARCHAR(11), chain VARCHAR(8), logpref VARCHAR(24), len INT(5)) ENGINE=MYISAM; --  create user rsyslogger@localhost identified by '...'; grant all privileges on traflog.* to rsyslogger@localhost; 

A tabela está pronta, o usuário está.

Agora adicione um gatilho, ele fará o que o logger falhou ao separar o endereço da porta, limpar os nomes das interfaces, remover os colchetes da bandeira:

 --   traffic DELIMITER // create TRIGGER delim_ip_port_flags BEFORE insert ON traffic FOR EACH ROW begin set NEW.inif = REGEXP_REPLACE ((NEW.inif), 'in:', '' ); set NEW.outif = REGEXP_REPLACE ((NEW.outif), 'out:', '' ); set NEW.sport = REGEXP_REPLACE ((NEW.src), '([0-9]+\.){3}[0-9]+:|([0-9]+\.){3}[0-9]+', '' ); set NEW.src = REGEXP_REPLACE ((NEW.src), ':[0-9]+', '' ); set NEW.dport = REGEXP_REPLACE ((NEW.dst), '([0-9]+\.){3}[0-9]+:|([0-9]+\.){3}[0-9]+', '' ); set NEW.dst = REGEXP_REPLACE ((NEW.dst), ':[0-9]+', '' ); set NEW.flags = REGEXP_REPLACE ((NEW.flags), '\\(|\\)', '' ); end // delimiter ; 

REGEXP_REPLACE procura o segundo parâmetro após o ponto decimal (temporada regular) e o substitui pelo terceiro parâmetro; no nosso caso, não há nada entre aspas; portanto, ele simplesmente remove o que encontra.

Vamos fazer uma inserção de teste, semelhante à maneira como o logger fará:

 --   insert into traffic (datetime, inif, outif, src, dst, smac, proto, chain, logpref) values (20180730075437, 'in:ether6', 'out:VLAN55', '192.168.0.234:4997', '192.168.6.18:65535', '00:15:17:31:b8:d7', 'TCP', '(SYN)', 'forward', 'BLOCKSMKNETS'); 

Vamos ver o que aconteceu:

 select * from tarffic; 

Se tudo estiver correto, siga em frente. Caso contrário, procure o erro.

Adicione pelo menos um índice. Eu não sou mestre em criar índices, mas como eu o entendo, no mysql é mais correto usar índices com campos de junção diferentes para consultas diferentes, pois uma consulta pode usar apenas um índice (ou estou errado?). Se você entender, faça-o a seu critério.
Muitas vezes tenho que fazer solicitações com um prefixo específico, então adicionei este índice:
 --  create index traffic_index on traffic (datetime,logpref,src); 

Feito.

Agora você precisa começar a enviar o roteador, adicionar as configurações e ações do servidor de log remoto, adicionar a opção de log a uma das regras de firewall, adicionar um prefixo de no máximo 24 caracteres.

No console Mikrotik, é algo como isto:

 /system logging action set 3 remote=192.168.0.94 src-address=192.168.0.230 add name=remote2 remote=192.168.0.19 syslog-facility=local6 target=remote /system logging add action=remote topics=error,account,critical,event,info add action=remote2 topics=firewall /ip firewall filter ... add action=drop chain=input comment="drop ssh brute forcers" dst-port=22,8291 log=yes log-prefix=DROP_SSH_BRUTE protocol=tcp src-address-list=ssh_blacklist ... 

Onde 192.168.0.230 é o endereço do roteador, 192.168.0.19 é o endereço do servidor de log para os logs do firewall e 192.168.0.94 é outro servidor de log, eu tenho os logs do sistema Mikrotik, não precisamos dele agora. Nossa configuração é remota2.

Em seguida, veja o que se enquadra no arquivo:

 tail -f /var/log/remote/192.168.0.230.log 

As linhas do roteador devem ser espalhadas no arquivo, a menos que sua regra funcione com bastante frequência.

Se alguns campos estiverem ausentes, ou seja, a sequência datetime, inif, outif, src, dst, smac, proto, flags, chain, logpref, len não for seguida, você poderá tentar alterar o parâmetro nos modelos de depuração do logger, substitua BLANK por DLFT. Então, em vez do vazio de qualquer campo, algumas letras aparecerão, não lembro quais. Se isso acontecer, algo está errado com a programação regular e você precisará corrigi-la.

Se tudo correu como deveria, desative as condições de teste e o modelo.

Você também precisa executar a configuração padrão em /etc/rsyslog.d/ abaixo, renomei-a para 50-default.conf para que os logs remotos não sejam despejados no log do sistema / var / log / message
Reinicie o logger.

Vamos esperar um pouco até que nosso banco de dados esteja cheio. Então podemos começar a seleção.

Algumas consultas para um exemplo:

Para ver o tamanho do banco de dados e o número de linhas:
 MariaDB [traflog]> select table_schema as "database", round(sum(data_length + index_length)/1024/1024,2) as "size Mb", TABLE_ROWS as "count rows" from information_schema.tables group by table_schema; +--------------------+---------+------------+ | database | size Mb | count rows | +--------------------+---------+------------+ | information_schema | 0.17 | NULL | | traflog | 3793.39 | 21839553 | +--------------------+---------+------------+ 2 rows in set (0.48 sec) 

Quase 4 GB cresceram em um mês, mas isso depende do número e das propriedades das regras de firewall registradas

Número de prefixos registrados
O número de prefixos registrados não é igual ao número de regras, algumas regras funcionam com um prefixo, mas ainda quantos prefixos totais? e quantas regras foram elaboradas para eles?

 MariaDB [traflog]> select logpref,count(logpref) from traffic group by logpref order by count(logpref) desc; +----------------------+----------------+ | logpref | count(logpref) | +----------------------+----------------+ | ACCEPT_TORF_INET | 14582602 | | ACCEPT_SMK_PPP | 1085791 | | DROP_FORWARD_INVALID | 982374 | | REJECT_BNK01 | 961503 | | ACCEPT_MMAX_TORF | 802455 | | ACCEPT_TORF_PPP | 736803 | | SMTP_DNAT | 689533 | | ACCEPT_SMK_INET | 451411 | | ACCEPT_INET_TORF | 389857 | | BLOCK_SMKNETS | 335424 | | DROP_SMTP_BRUTE | 285850 | | ACCEPT_ROZN_TORF | 154811 | | ACCEPT_TORF_MMAX | 148393 | | DROP_ETHALL_ETHALL | 80679 | | ACCEPT_SMTP | 48921 | | DROP_SMTP_DDOS | 32190 | | RDP_DNAT | 28757 | | ACCEPT_TORF_ROZN | 18456 | | SIP_DNAT | 15494 | | 1CWEB_DNAT | 6406 | | BLOCKSMKNETS | 5789 | | DROP_SSH_BRUTE | 3162 | | POP_DNAT | 1997 | | DROP_RDP_BRUTE | 442 | | DROP_BNK01 | 291 | | DROPALL | 138 | | ACCEPT_RTP_FORWARD | 90 | | REJECT_SMTP_BRUTE | 72 | | L2TP_INPUT_ACCEPT | 33 | +----------------------+----------------+ 29 rows in set (2 min 51.03 sec) 

ACCEPT_TORF_INET é o líder, por esse prefixo, você pode encontrar todos os que acessaram a Internet a partir de nossa rede local, protocolos e portas são registrados, chegará a hora e o acesso será fechado para alguns. Existem dados de referência para trabalhos futuros sobre bugs.

Líder da lança Smtp
Vamos ver quem hoje tentou acessar o servidor smtp:

 MariaDB [traflog]> select src,count(dport) from traffic where logpref='SMTP_DNAT' and datetime > '2018101600000000' group by src order by count(dport) desc limit 10; +----------------+--------------+ | src | count(dport) | +----------------+--------------+ | 191.96.249.92 | 12440 | | 191.96.249.24 | 4556 | | 191.96.249.61 | 4537 | | 185.255.31.122 | 3119 | | 178.57.79.250 | 226 | | 185.36.81.174 | 216 | | 185.234.219.32 | 211 | | 89.248.162.145 | 40 | | 45.125.66.157 | 32 | | 188.165.124.31 | 21 | +----------------+--------------+ 10 rows in set, 1 warning (21.36 sec) 

Claramente, o nó 191.96.249.92 é o vencedor hoje. Vamos ver em quais regras registradas ele ainda apareceu:

 MariaDB [traflog]> select src,dport,count(dport),logpref from traffic where src='191.96.249.92' group by logpref order by count(dport) desc; +---------------+-------+--------------+-----------------+ | src | dport | count(dport) | logpref | +---------------+-------+--------------+-----------------+ | 191.96.249.92 | 25 | 226989 | SMTP_DNAT | | 191.96.249.92 | 25 | 170714 | DROP_SMTP_BRUTE | | 191.96.249.92 | 25 | 2907 | DROP_SMTP_DDOS | | 191.96.249.92 | 25 | 2061 | ACCEPT_SMTP | +---------------+-------+--------------+-----------------+ 4 rows in set (10 min 44.21 sec) 

Este é especializado apenas em smtp, ~ 1% dos acessos por tentar adivinhar a senha ou tentar enviar um pouco de lixo; o restante foi para a casa de banhos.

A solicitação demorou 10 minutos, é muito, os índices atuais não são adequados para ela ou você pode reformular a solicitação, mas agora não falaremos sobre isso.

No futuro, está planejado estragar a interface da web com consultas e formulários padrão.
O vetor é dado, espero que este artigo seja útil.

Obrigado a todos!

Referências:

Documentação Rsyslog
Documentação do mysql
Documentação de log do Mikrotik

Obrigado à comunidade LOR pelas dicas.

UPD.1
Adicionado o campo de sinalizadores ao banco de dados, agora você pode acompanhar a duração da conexão capturando SYN, FIN.
Corrigidos alguns erros nos regulares do rsyslog, bem como nos gatilhos do mysql.

Curiosamente, a regra padrão defconf: drop invalid remove todos os pacotes finais de conexões TCP, como resultado, todos os nós que tentam fechar a conexão na ciência falham, enviando vários FINs. Isso está certo?

Eu adicionei uma regra que permite a travessia do TCP com sinalizadores ACK, FIN.

Sob o spoiler SQL, um procedimento que mostra o tempo das conexões TCP nos últimos cinco minutos
connections_list ()
 DROP PROCEDURE IF EXISTS connections_list; DELIMITER // CREATE PROCEDURE connections_list() BEGIN DECLARE logid BIGINT UNSIGNED; DECLARE done INT DEFAULT FALSE; DECLARE datefin DATETIME; DECLARE datesyn DATETIME; DECLARE conntime TIME; DECLARE connsport INT; DECLARE conndport INT; DECLARE connsrc VARCHAR(21); DECLARE conndst VARCHAR(21); DECLARE cur CURSOR FOR SELECT id,datetime,src,sport,dst,dport FROM conn_syn_fin WHERE flags='SYN'; DECLARE CONTINUE HANDLER FOR NOT FOUND SET done=TRUE; DROP TABLE IF EXISTS conn_syn_fin; DROP TABLE IF EXISTS connless; CREATE temporary TABLE connless(datestart DATETIME,dateend DATETIME,duration TIME,src VARCHAR(21),sport INT,dst VARCHAR(21),dport INT); CREATE temporary TABLE conn_syn_fin (SELECT * from traffic WHERE datetime > now() - interval 5 minute and src in (select src from traffic where datetime > now() - interval 5 minute and logpref='TCP_FIN' and flags like '%FIN%') and (flags like '%SYN%' or flags like '%FIN%') order by id); OPEN cur; read_loop: LOOP FETCH cur INTO logid,datesyn,connsrc,connsport,conndst,conndport; IF done THEN LEAVE read_loop; END IF; set datefin=(SELECT datetime FROM conn_syn_fin WHERE id>logid and src=connsrc and sport=connsport and flags like '%FIN%' and dst=conndst and dport=conndport limit 1); set conntime=(SELECT timediff(datefin,datesyn)); INSERT INTO connless (datestart,dateend,duration,src,sport,dst,dport) value (datesyn,datefin,conntime,connsrc,connsport,conndst,conndport); END LOOP; CLOSE cur; select * from connless; END; // DELIMITER ; 


Como resultado do procedimento, duas tabelas temporárias serão criadas.
A tabela conn_syn_fin contém entradas de log com os sinalizadores SYN e FIN e, em seguida, é realizada uma pesquisa usando o cursor nesta tabela.
A tabela sem conexão contém uma lista de conexões, abertas e concluídas, concluídas com uma duração de abertura, respectivamente, não.
Observe o tempo de amostragem menos cinco minutos a partir da hora atual. Meu pedido é lento. Lentamente, ele percorre a pesquisa do cursor, processa cerca de 10 registros por segundo, tenta acelerar de todas as formas, mas o tempo de execução é sempre o mesmo.
Observe também que este procedimento é apenas para fins de demonstração. Se você precisar fazer uma seleção para um src / sport / dst / dport específico, é melhor fazer um procedimento separado semelhante a este. Se você é um mestre sql, pode escrever sua consulta melhor.

chame connections_list ();
 MariaDB [traflog]> call connections_list(); +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ | datestart | dateend | duration | src | sport | dst | dport | +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ | 2019-03-20 14:12:19 | 2019-03-20 14:13:14 | 00:00:55 | 192.168.0.81 | 41868 | 87.250.250.207 | 443 | | 2019-03-20 14:12:25 | NULL | NULL | 192.168.0.65 | 49311 | 52.5.23.125 | 443 | | 2019-03-20 14:12:31 | 2019-03-20 14:12:51 | 00:00:20 | 192.168.0.104 | 54433 | 217.69.139.42 | 443 | | 2019-03-20 14:12:31 | 2019-03-20 14:12:51 | 00:00:20 | 192.168.0.104 | 54434 | 217.69.139.42 | 443 | | 2019-03-20 14:12:32 | NULL | NULL | 192.168.0.119 | 37977 | 209.85.233.95 | 443 | ... | 2019-03-20 14:17:12 | NULL | NULL | 192.168.0.119 | 39331 | 91.213.158.131 | 443 | | 2019-03-20 14:17:13 | NULL | NULL | 192.168.0.90 | 63388 | 87.240.185.236 | 443 | +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ 399 rows in set (33.17 sec) Query OK, 0 rows affected (33.18 sec) 



Após a conclusão do procedimento, as tabelas temporárias conn_syn_fin e connless permanecem, você pode vê-las com mais detalhes se encontrar algo suspeito ou não confiável. Após o início do procedimento, as tabelas antigas são excluídas e novas são exibidas. Escreva se encontrar algum erro.

Source: https://habr.com/ru/post/pt427159/


All Articles