zeek2es.py

Questa applicazione Python traduce i log di Zeek in formato ASCII TSV e JSON nel formato di caricamento bulk JSON di ElasticSearch.

Indice:

• Introduzione
• Installazione
  • Elastic v8.0+
• Aggiornare zeek2es
  • Pipeline di Ingest ES
• Filtraggio dei dati
  • Filtri Python
  • Filtro sulle chiavi
• Esempi da riga di comando
• Opzioni della riga di comando
• Requisiti
• Note
  • Humio
  • Input di log JSON
  • Data Streams
  • Script ausiliari
  • Cython

Introduzione

Questa applicazione riconosce i log gzip o non compressi. Questa applicazione presuppone che ElasticSearch sia configurato sul tuo localhost sulla porta predefinita. Se non hai ElasticSearch puoi inviare il JSON a stdout con le opzioni da riga di comando -s -b per elaborarlo con l'applicazione jq.

Puoi aggiungere un sottocampo keyword ai campi di testo con l'opzione da riga di comando -k. Questo è utile per le aggregazioni in Kibana.

Se Python è già presente sul tuo sistema, non c'è nient'altro da copiare sulla tua macchina oltre a Elasticsearch, Kibana e zeek2es.py se hai già installata la libreria requests.

Installazione

Supponendo che tu soddisfi i requisiti, non è necessaria alcuna installazione. Devi solo copiare zeek2es.py sul tuo host ed eseguirlo con Python. Una volta che i log di Zeek sono stati importati con la generazione automatica del nome dell'indice (cioè, se non hai fornito l'opzione -i) troverai i tuoi indici denominati "zeek_zeeklogname_date", dove zeeklogname è un nome di log come conn e date è nel formato YYYY-MM-DD. In questo caso, imposta il pattern dell'indice Kibana in modo che corrisponda a zeek*. Se hai nominato il tuo indice con l'opzione -i, dovrai creare un pattern di indice Kibana che corrisponda al tuo schema di denominazione.

Se stai aggiornando zeek2es, consulta la sezione sull'aggiornamento di zeek2es.

Elastic v8.0+

Dovrai anche modificare i comandi curl in questo readme per includere `-k -u elastic:<password>`
dove la password dell'utente `elastic` è impostata con un comando come il seguente:```
./bin/elasticsearch-reset-password -u elastic -i

Docker

Ora puoi inserire i log nella directory `VOLUME_MOUNT/data/logs` (`VOLUME_MOUNT` che imposti nel file `.env`).
Quando i log vengono CREATI in questa directory, zeek2es inizierà a elaborarli e a inviarli in Elasticsearch.
Puoi quindi accedere a https://localhost:5601 con il nome utente e la password configurati nel file `.env`.  
Per impostazione predefinita è presente un certificato autofirmato, ma puoi cambiarlo se modifichi i file docker compose.  Una volta dentro
Kibana andrai su Stack Management->Data Views e creerai una data view per `logs*` con il timestamp `@timestamp`.
Ora potrai andare su Discover e iniziare a cercare i tuoi log!  I tuoi dati sono persistenti nella directory `VOLUME_MOUNT/data` che hai impostato.
Se desideri rimuovere tutti i dati, esegui semplicemente `rm -rf VOLUME_MOUNT/data`, sostituendo la directory che hai impostato in quel comando di rimozione.
La prossima volta che avvierai il cluster sarà completamente nuovo per ulteriori dati.

## Aggiornare zeek2es <a name="upgradingzeek2es" />

La maggior parte degli aggiornamenti dovrebbe essere semplice come copiare il [zeek2es.py](https://github.com/corelight/zeek2es/blob/HEAD/zeek2es.py) più recente sopra 
quello vecchio.  In alcuni casi, la pipeline di ingest ES richiesta per l'opzione da riga di comando `-g` 
potrebbe cambiare durante un aggiornamento.  Pertanto, si consiglia vivamente di eliminare 
la propria [pipeline di ingest](#esingestpipeline) prima di eseguire una nuova versione di zeek2es.py.

### Pipeline di Ingest ES <a name="esingestpipeline" />

Se devi [eliminare la pipeline di ingest ES "zeekgeoip"](https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-pipeline-api.html) 
usata per geolocalizzare gli indirizzi IP con l'opzione da riga di comando `-g`, puoi farlo graficamente
tramite Stack Management->Ingest Pipelines di Kibana oppure questo comando lo farà per te:```
curl -X DELETE "localhost:9200/_ingest/pipeline/zeekgeoip?pretty"

Filtraggio dei Dati

Filtri Python

O forse preferisci filtrare le connessioni che hanno almeno 1.024 byte, con almeno 1 byte proveniente dalla destinazione:```
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

Per gli utenti esperti, l'opzione `-f` consente di definire una funzione completa (invece delle funzioni lambda di Python) così da poter scrivere funzioni che si estendono su più righe.

### Filtro sulle chiavi <a name="filteronkeys" />

In alcuni casi potresti voler estrarre dati da un log che dipendono da un altro. Un esempio potrebbe essere trovare tutte le righe di `ssl.log` che hanno un `uid` corrispondente a righe precedentemente indicizzate da `conn.log`, o viceversa. Puoi filtrare importando i tuoi file `conn.log` con la riga di comando `-o uid uid.txt`. Questo registrerà tutti gli uid indicizzati in un file chiamato `uid.txt`. Poi, quando importi i tuoi file `ssl.log`, fornisci la riga di comando `-e uid uid.txt`. Questo importerà solo le righe SSL contenenti valori `uid` presenti in `uid.txt`, precedentemente creato dalla nostra importazione di `conn.log`.

## Esempi di riga di comando <a name="commandlineexamples" />```
python zeek2es.py your_zeek_log.gz -i your_es_index_name

Poiché gli indici hanno la data aggiunta, potresti eliminare il 31 dicembre 2021 con il seguente comando:``` curl -X DELETE http://localhost:9200/zeek_*_2021-12-31

Potresti eliminare tutte le voci di conn.log con questo comando:```
curl -X DELETE http://localhost:9200/zeek_conn_*

Opzioni della riga di 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" />

- Un ambiente Unix-like (MacOs funziona!)
- Python
  - Libreria Python [requests](https://docs.python-requests.org/en/latest/) installata, ad esempio con `pip`.

## Notes <a name="notes" />

### Humio <a name="humio" />

Per importare i tuoi dati in Humio dovrai configurare un repository con il parser `corelight-json`. Ottieni
il token di ingestione per il repository e potrai importare i tuoi dati con un comando come:```
python3 zeek2es.py -s -b --humio http://localhost:8080 b005bf74-1ed3-4871-904f-9460a4687202 http.log 

L'URL deve essere nel formato: http://yourserver:8080, poiché il resto del percorso viene aggiunto dallo script zeek2es.py automaticamente per te.

JSON Log Input

Flussi di dati

Puoi utilizzare i flussi di dati invece degli indici per log di grandi dimensioni con l'opzione -d della riga di comando. Questa opzione crea modelli di indice che iniziano con zeek_. Crea anche una politica del ciclo di vita chiamata zeek-lifecycle-policy. Se desideri eliminare tutti i flussi di dati, le politiche del ciclo di vita, e i modelli di indice, questi comandi lo faranno per te:``` 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

### Script di supporto <a name="helperscripts" />

Ci sono due script che ti aiuteranno a trasformare i tuoi log in flussi di dati come `logs-zeek-conn`.
Il primo script è [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/HEAD/process_logs_as_datastream.sh) e, data 
una lista di log e directory, li importa come tali.  Il secondo script 
è [process_log.sh](https://github.com/corelight/zeek2es/blob/HEAD/process_log.sh) e può essere usato per importare i log 
uno alla volta.  Questo script può anche essere usato per monitorare i log creati in una directory con 
[fswatch](https://emcrisostomo.github.io/fswatch/).  Entrambi gli script hanno righe di comando di esempio 
se li esegui senza alcun parametro.```
$ ./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

Lo strumento fornisce il payload JavaScript finale che riflette l'intero output del canvas, consentendoti di visualizzare e analizzare rapidamente la rappresentazione visiva dell'URL di destinazione all'interno dei dati acquisiti.``` $ ./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 {} "" :::: -

Dovrai modificare questi script e le righe di comando in base al tuo ambiente.  

Qualsiasi file avente un nome di log come `conn_filter.txt` nella `lambda_filter_file_dir`, per impostazione predefinita la tua directory home, verrà applicato come file di filtro lambda
all'input di log corrispondente.  Questo ti consente di impostare tutti i tuoi filtri in un'unica directory e importare più file di log con
quel set di filtri in un unico comando con [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/HEAD/process_logs_as_datastream.sh).

Le seguenti righe dovrebbero eliminare tutti i dati Zeek in ElasticSearch, indipendentemente dal fatto che si utilizzino indici o 
data stream, o questi script di supporto:```
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

... o se si utilizza 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

Ma per poter fare questo in v8+ dovrai configurare Elastic come descritto 
nella sezione [Elastic v8.0+](#elastic80).

### Cython <a name="cython" />

Se desideri provare [Cython](https://cython.org/), devi eseguire `python setup.py build_ext --inplace` 
per generare il tuo file compilato.  Devi farlo ogni volta che aggiorni zeek2es!