
Uma aplicação Python para filtrar e transferir logs do Zeek para o Elastic/OpenSearch+Humio. Esta aplicação também pode emitir logs JSON puros para o stdout para processamento posterior!
Esta aplicação Python converte os logs ASCII TSV e JSON do Zeek para o formato JSON de carga em massa do ElasticSearch.

Quer ver vários logs do Zeek para o mesmo ID de conexão (uid) ou ID de arquivo (fuid)? Aqui estão as ocorrências de files.log, http.log e conn.log para um único uid:

Você pode realizar buscas por sub-rede no tipo 'addr' do Zeek:

Você pode criar gráficos de séries temporais, como este gráfico de NTP e HTTP:

Endereços IP podem ser geolocalizados com a opção de linha de comando -g:

Agregações são simples e rápidas:

Esta aplicação "simplesmente funcionará" quando os formatos de log do Zeek mudarem. A lógica lê os nomes dos campos e os tipos associados para configurar os mapeamentos corretamente no ElasticSearch.
Esta aplicação reconhecerá logs gzip ou descompactados. Esta aplicação presume
que você tenha o ElasticSearch configurado no seu localhost na porta padrão.
Se você não tiver o ElasticSearch, pode enviar o JSON para o stdout com as opções de linha de comando -s -b
para processar com a aplicação jq.
Você pode adicionar um subcampo keyword aos campos de texto com a opção de linha de comando -k. Isso é útil
para agregações no Kibana.
Se o Python já estiver no seu sistema, não há nada adicional para você copiar para a sua máquina além de Elasticsearch, Kibana e zeek2es.py se você já tiver a biblioteca requests instalada.
Supondo que você atenda aos requisitos, não há instalação. Basta
copiar o zeek2es.py para o seu host e executá-lo com Python. Depois que os logs do Zeek
forem importados com a geração automática de nome de índice (ou seja, você não forneceu a opção -i),
você encontrará seus índices nomeados como "zeek_zeeklogname_date", onde zeeklogname é um nome de log como conn
e a date está no formato YYYY-MM-DD. Defina seu padrão de índice no Kibana para corresponder a zeek* neste caso. Se
você nomeou seu índice com a opção -i, precisará criar um padrão de índice no Kibana que
corresponda ao seu esquema de nomenclatura.
Se você está atualizando o zeek2es, consulte a seção sobre atualização do zeek2es.
Como os índices têm a data anexada a eles, você poderia excluir 31 de dezembro de 2021 com o seguinte comando:``` curl -X DELETE http://localhost:9200/zeek_*_2021-12-31
Você poderia excluir todas as entradas do conn.log com este comando:```
curl -X DELETE http://localhost:9200/zeek_conn_*
optional arguments:
-h, --help show this help message and exit
-i ESINDEX, --esindex ESINDEX
The Elasticsearch index/data stream name.
-u ESURL, --esurl ESURL
The Elasticsearch URL. Use ending slash. Use https for Elastic v8+. (default: http://localhost:9200)
--user USER The Elasticsearch user. (default: disabled)
--passwd PASSWD The Elasticsearch password. Note this will put your password in this shell history file. (default: disabled)
-l LINES, --lines LINES
Lines to buffer for RESTful operations. (default: 10,000)
-n NAME, --name NAME The name of the system to add to the index for uniqueness. (default: empty string)
-k KEYWORDS [KEYWORDS ...], --keywords KEYWORDS [KEYWORDS ...]
A list of text fields to add a keyword subfield. (default: service)
-a LAMBDAFILTER, --lambdafilter LAMBDAFILTER
A Python lambda function, when eval'd will filter your output JSON dict. (default: empty string)
-f FILTERFILE, --filterfile FILTERFILE
A Python function file, when eval'd will filter your output JSON dict. (default: empty string)
-y OUTPUTFIELDS [OUTPUTFIELDS ...], --outputfields OUTPUTFIELDS [OUTPUTFIELDS ...]
A list of fields to keep for the output. Must include ts. (default: empty string)
-d DATASTREAM, --datastream DATASTREAM
Instead of an index, use a data stream that will rollover at this many GB.
Recommended is 50 or less. (default: 0 - disabled)
--compress If a datastream is used, enable best compression.
-o fieldname filename, --logkey fieldname filename
A field to log to a file. Example: uid uid.txt.
Will append to the file! Delete file before running if appending is undesired.
This option can be called more than once. (default: empty - disabled)
-e fieldname filename, --filterkeys fieldname filename
A field to filter with keys from a file. Example: uid uid.txt. (default: empty string - disabled)
-g, --ingestion Use the ingestion pipeline to do things like geolocate IPs and split services. Takes longer, but worth it.
-p SPLITFIELDS [SPLITFIELDS ...], --splitfields SPLITFIELDS [SPLITFIELDS ...]
A list of additional fields to split with the ingestion pipeline, if enabled.
(default: empty string - disabled)
-j, --jsonlogs Assume input logs are JSON.
-r, --origtime Keep the numerical time format, not milliseconds as ES needs.
-t, --timestamp Keep the time in timestamp format.
-s, --stdout Print JSON to stdout instead of sending to Elasticsearch directly.
-b, --nobulk Remove the ES bulk JSON header. Requires --stdout.
--humio HUMIO HUMIO First argument is the Humio URL, the second argument is the ingest token.
-c, --cython Use Cython execution by loading the local zeek2es.so file through an import.
Run python setup.py build_ext --inplace first to make your zeek2es.so file!
-w, --hashdates Use hashes instead of dates for the index name.
-z, --supresswarnings
Supress any type of warning. Die stoically and silently.
To delete indices:
curl -X DELETE http://localhost:9200/zeek*?pretty
To delete data streams:
curl -X DELETE http://localhost:9200/_data_stream/zeek*?pretty
To delete index templates:
curl -X DELETE http://localhost:9200/_index_template/zeek*?pretty
To delete the lifecycle policy:
curl -X DELETE http://localhost:9200/_ilm/policy/zeek-lifecycle-policy?pretty
You will need to add -k -u elastic_user:password if you are using Elastic v8+.
## Requirements <a name="requirements" />
- A Unix-like environment (MacOs works!)
- Python
- [requests](https://docs.python-requests.org/en/latest/) Python library installed, such as with with `pip`.
## Notes <a name="notes" />
### Humio <a name="humio" />
To import your data into Humio you will need to set up a repository with the `corelight-json` parser. Obtain
the ingest token for the repository and you can import your data with a command such as:```
python3 zeek2es.py -s -b --humio http://localhost:8080 b005bf74-1ed3-4871-904f-9460a4687202 http.log
A URL deve estar no formato: http://yourserver:8080, pois o restante do caminho é adicionado pelo
script zeek2es.py automaticamente para você.
Você pode usar data streams em vez de índices para logs grandes com a opção de linha de comando -d. Esta
opção cria modelos de índice que começam com zeek_. Ela também cria uma política de ciclo de vida
chamada zeek-lifecycle-policy. Se você quiser excluir todos os seus data streams, políticas de ciclo de vida
e modelos de índice, estes comandos farão isso por você:```
curl -X DELETE http://localhost:9200/_data_stream/zeek*?pretty
curl -X DELETE http://localhost:9200/_index_template/zeek*?pretty
curl -X DELETE http://localhost:9200/_ilm/policy/zeek-lifecycle-policy?pretty
### Scripts Auxiliares <a name="helperscripts" />
Existem dois scripts que ajudarão você a transformar seus logs em streams de dados, como `logs-zeek-conn`.
O primeiro script é [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh) e, dada
uma lista de logs e diretórios, os importará como tal. O segundo script
é [process_log.sh](https://github.com/corelight/zeek2es/blob/master/process_log.sh), e pode ser usado para importar logs
um de cada vez. Este script também pode ser usado para monitorar logs criados em um diretório com
[fswatch](https://emcrisostomo.github.io/fswatch/). Ambos os scripts têm linhas de comando de exemplo
se você os executar sem nenhum parâmetro.```
$ ./process_logs_as_datastream.sh
Usage: ./process_logs_as_datastream.sh NJOBS "ADDITIONAL_ARGS_TO_ZEEK2ES" "LIST_OF_LOGS_DELIMITED_BY_SPACES" DIR1 DIR2 ...
Example:
time ./process_logs_as_datastream.sh 16 "" "amqp bgp conn dce_rpc dhcp dns dpd files ftp http ipsec irc kerberos modbus modbus_register_change mount mqtt mysql nfs notice ntlm ntp ospf portmap radius reporter rdp rfb rip ripng sip smb_cmd smb_files smb_mapping smtp snmp socks ssh ssl stun syslog tunnel vpn weird wireguard x509" /usr/local/var/logs
I don't see any text to translate — the INPUT section is empty. Please provide the chunk content so I can translate it from English to Portuguese.``` $ ./process_log.sh Usage: ./process_log.sh LOGFILENAME "ADDITIONAL_ARGS_TO_ZEEK2ES"
Example: fswatch -m poll_monitor --event Created -r /data/logs/zeek | awk '/^./(conn|dns|http)...log.gz$/' | parallel -j 16 ./process_log.sh {} "" :::: -
Você precisará editar esses scripts e linhas de comando de acordo com seu ambiente.
Quaisquer arquivos que tenham nome de log, como `conn_filter.txt` no `lambda_filter_file_dir`, por padrão seu diretório inicial, serão aplicados como um arquivo de filtro lambda à entrada de log correspondente. Isso permite configurar todos os seus filtros em um diretório e importar vários arquivos de log com
esse conjunto de filtros em um único comando com [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh).
As linhas a seguir devem excluir todos os dados do Zeek no ElasticSearch, não importa se você usa índices ou
streams de dados, ou estes scripts auxiliares:```
curl -X DELETE http://localhost:9200/zeek*?pretty
curl -X DELETE http://localhost:9200/_data_stream/zeek*?pretty
curl -X DELETE http://localhost:9200/_data_stream/logs-zeek*?pretty
curl -X DELETE http://localhost:9200/_index_template/zeek*?pretty
curl -X DELETE http://localhost:9200/_index_template/logs-zeek*?pretty
curl -X DELETE http://localhost:9200/_ilm/policy/zeek-lifecycle-policy?pretty
... ou se estiver usando Elastic v8+ ...``` curl -X DELETE -k -u elastic:password https://localhost:9200/zeek*?pretty curl -X DELETE -k -u elastic:password https://localhost:9200/_data_stream/zeek*?pretty curl -X DELETE -k -u elastic:password https://localhost:9200/_data_stream/logs-zeek*?pretty curl -X DELETE -k -u elastic:password https://localhost:9200/_index_template/zeek*?pretty curl -X DELETE -k -u elastic:password https://localhost:9200/_index_template/logs-zeek*?pretty curl -X DELETE -k -u elastic:password https://localhost:9200/_ilm/policy/zeek-lifecycle-policy?pretty
Mas para conseguir fazer isso no v8+ você precisará configurar o Elastic conforme descrito
na seção [Elastic v8.0+](#elastic80).
### Cython <a name="cython" />
Se você quiser experimentar o [Cython](https://cython.org/), você deve executar `python setup.py build_ext --inplace`
primeiro para gerar o arquivo compilado. Você deve fazer isso toda vez que atualizar o zeek2es!