zeek2es.py

Esta aplicação Python converte os logs ASCII TSV e JSON do Zeek para o formato JSON de carga em massa do ElasticSearch.

Índice:

• Introdução
• Instalação
  • Elastic v8.0+
  • Docker
• Atualizando o zeek2es
  • Pipeline de Ingestão ES
• Filtrando Dados
  • Filtros Python
  • Filtrar por Chaves
• Exemplos de Linha de Comando
• Opções de Linha de Comando
• Requisitos
• Notas
  • Humio
  • Entrada de Log JSON
  • Fluxos de Dados
  • Scripts Auxiliares
  • Cython

Introdução

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.

Instalação

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.

Elastic v8.0+

Você também precisará alterar os comandos curl neste readme para conter `-k -u elastic:<password>`
onde a senha do usuário `elastic` é definida com um comando como o seguinte:```
./bin/elasticsearch-reset-password -u elastic -i

Docker

Agora você pode colocar logs no diretório `VOLUME_MOUNT/data/logs` (`VOLUME_MOUNT` que você definiu no arquivo `.env`).
Quando logs são CRIADOS neste diretório, o zeek2es começará a processá-los e enviá-los para o Elasticsearch.
Em seguida, você pode fazer login em https://localhost:5601 com o nome de usuário e a senha que você configurou no arquivo `.env`.  
Por padrão, há um certificado autoassinado, mas você pode alterá-lo se editar os arquivos do docker compose.  Uma vez dentro
do Kibana, você irá para Stack Management->Data Views e criará uma data view para `logs*` com o timestamp `@timestamp`.
Agora você poderá ir para o Discover e começar a pesquisar seus logs!  Seus dados são persistentes no diretório `VOLUME_MOUNT/data` que você definiu.
Se você quiser remover todos os dados, basta `rm -rf VOLUME_MOUNT/data`, substituindo o diretório que você definiu nesse comando de remoção.
Na próxima vez que você iniciar seu cluster, ele estará novinho em folha para mais dados.

## Atualizando o zeek2es <a name="upgradingzeek2es" />

A maioria das atualizações deve ser tão simples quanto copiar o novo [zeek2es.py](https://github.com/corelight/zeek2es/blob/HEAD/zeek2es.py) sobre 
o antigo.  Em alguns casos, o pipeline de ingestão do ES necessário para a opção de linha de comando `-g` 
pode mudar durante uma atualização.  Portanto, é fortemente recomendado que você exclua 
seu [pipeline de ingestão](#esingestpipeline) antes de executar uma nova versão do zeek2es.py.

### Pipeline de Ingestão do ES <a name="esingestpipeline" />

Se você precisar [excluir o pipeline de ingestão do ES "zeekgeoip"](https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-pipeline-api.html) 
usado para geolocalizar endereços IP com a opção de linha de comando `-g`, você pode fazer isso graficamente
por meio do Stack Management->Ingest Pipelines do Kibana ou este comando fará isso por você:```
curl -X DELETE "localhost:9200/_ingest/pipeline/zeekgeoip?pretty"

Filtering Data

Python Filters

Ou talvez você queira filtrar conexões que tenham pelo menos 1,024 bytes, com pelo menos 1 byte vindo do 
destino:```
lambda x: 'orig_ip_bytes' in x and 'resp_ip_bytes' in x and x['orig_ip_bytes'] + x['resp_ip_bytes'] > 1024 and x['resp_ip_bytes'] > 0

Para usuários avançados, a opção `-f` permitirá que você defina uma função completa (em vez das funções lambda do Python) para que você possa escrever funções que 
se estendam por várias linhas.

### Filtrar por Chaves <a name="filteronkeys" />

Em alguns casos, você pode querer extrair dados de um log que dependem de outro.  Um
exemplo seria encontrar todas as linhas de `ssl.log` que tenham um `uid` correspondente a linhas previamente
indexadas de `conn.log`, ou vice-versa.  Você pode filtrar importando seus
arquivos `conn.log` com a linha de comando `-o uid uid.txt`.  Isso registrará todos os uids que foram 
indexados em um arquivo chamado `uid.txt`.  Depois, ao importar seus arquivos `ssl.log`, você fornecerá 
a linha de comando `-e uid uid.txt`.  Isso importará apenas as linhas SSL 
que contenham valores de `uid` presentes em `uid.txt`, criado anteriormente a partir da importação do `conn.log`.

## Exemplos de Linha de Comando <a name="commandlineexamples" />```
python zeek2es.py your_zeek_log.gz -i your_es_index_name

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_*

Opções de Linha de Comando ```

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ê.

Entrada de Logs JSON

Data Streams

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/HEAD/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/HEAD/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/HEAD/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!