corelight/zeek2esUne application Python pour filtrer et transférer les journaux Zeek vers Elastic/OpenSearch+Humio. Cette application peut également produire des journaux JSON purs sur stdout pour un traitement ultérieur !
Cette application Python traduit les journaux Zeek au format TSV ASCII et JSON dans le format de chargement en masse JSON d'ElasticSearch.
Vous voulez voir plusieurs journaux Zeek pour le même ID de connexion (uid) ou ID de fichier (fuid) ? Voici les résultats de files.log, http.log et conn.log pour un seul uid :
Vous pouvez effectuer des recherches de sous-réseau sur le type 'addr' de Zeek :
Vous pouvez créer des graphiques de séries chronologiques, comme ce graphique NTP et HTTP :
Les adresses IP peuvent être géolocalisées avec l'option de ligne de commande -g :
Les agrégations sont simples et rapides :
Cette application fonctionnera « simplement » lorsque les formats de journaux Zeek changeront. La logique lit les noms de champs et les types associés pour configurer correctement les mappings dans ElasticSearch.
Cette application reconnaîtra les journaux gzip ou non compressés. Cette application suppose
qu'ElasticSearch est configuré sur votre localhost au port par défaut.
Si vous n'avez pas ElasticSearch, vous pouvez sortir le JSON vers stdout avec les options de ligne de commande -s -b
pour le traiter avec l'application jq.
Vous pouvez ajouter un sous-champ keyword aux champs texte avec l'option de ligne de commande -k. Ceci est utile
pour les agrégations dans Kibana.
Si Python est déjà sur votre système, il n'y a rien de supplémentaire à copier sur votre machine que Elasticsearch, Kibana et zeek2es.py si vous avez déjà la bibliothèque requests installée.
En supposant que vous remplissez les prérequis, il n'y en a pas. Vous copiez simplement
zeek2es.py sur votre hôte et l'exécutez avec Python. Une fois que les journaux Zeek
ont été importés avec la génération automatique de noms d'index (c'est-à-dire que vous n'avez pas fourni l'option -i),
vous trouverez vos index nommés « zeek_nomdujournalzeek_date », où nomdujournalzeek est un nom de journal comme conn
et date est au format AAAA-MM-JJ. Définissez votre modèle d'index Kibana pour qu'il corresponde à zeek* dans ce cas.
Si vous avez nommé votre index avec l'option -i, vous devrez créer un modèle d'index Kibana qui
correspond à votre schéma de nommage.
Si vous mettez à niveau zeek2es, veuillez consulter la section sur la mise à niveau de zeek2es.
Comme les index ont la date ajoutée à leur nom, vous pourriez supprimer le 31 décembre 2021 avec la commande suivante :``` curl -X DELETE http://localhost:9200/zeek_*_2021-12-31
Vous pouvez supprimer toutes les entrées de conn.log avec cette commande :```
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+.
## Configuration requise <a name="requirements" />
- Un environnement de type Unix (MacOS fonctionne !)
- Python
- Bibliothèque Python [requests](https://docs.python-requests.org/en/latest/) installée, par exemple avec `pip`.
## Notes <a name="notes" />
### Humio <a name="humio" />
Pour importer vos données dans Humio, vous devrez configurer un référentiel avec l'analyseur `corelight-json`. Obtenez
le jeton d'ingestion pour le référentiel et vous pourrez importer vos données avec une commande telle que :```
python3 zeek2es.py -s -b --humio http://localhost:8080 b005bf74-1ed3-4871-904f-9460a4687202 http.log
L'URL doit être au format : http://yourserver:8080, car le reste du chemin est ajouté automatiquement par le script zeek2es.py pour vous.
Vous pouvez utiliser des flux de données au lieu d'index pour les journaux volumineux avec l'option de ligne de commande -d. Cette option crée des modèles d'index commençant par zeek_. Elle crée également une politique de cycle de vie nommée zeek-lifecycle-policy. Si vous souhaitez supprimer tous vos flux de données, politiques de cycle de vie et modèles d'index, ces commandes le feront pour vous :```
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 auxiliaires <a name="helperscripts" />
Il existe deux scripts qui vous aideront à transformer vos logs en flux de données tels que `logs-zeek-conn`.
Le premier script est [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh) et étant donné
une liste de logs et de répertoires, les importera en tant que tels. Le second script
est [process_log.sh](https://github.com/corelight/zeek2es/blob/master/process_log.sh), et il peut être utilisé pour importer les logs
un à la fois. Ce script peut également être utilisé pour surveiller les logs créés dans un répertoire avec
[fswatch](https://emcrisostomo.github.io/fswatch/). Les deux scripts ont des exemples de lignes de commande
si vous les exécutez sans paramètres.```
$ ./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
Plongée technique:
Protocoles de transport:
Configuration du serveur:
VShell utilise une approche de configuration avec:
VShellConfig.xml pour les paramètres globaux.Listes de contrôle d'accès (ACL):
Contrôles d'accès granulaires par utilisateur et par groupe, permettant:
Système de déclenchement:
VShell offre un mécanisme de déclenchement puissant pour des actions personnalisées lorsque des événements se produisent:
Déploiements de configuration courants:
VShell est généralement déployé comme:
Example: fswatch -m poll_monitor --event Created -r /data/logs/zeek | awk '/^./(conn|dns|http)...log.gz$/' | parallel -j 16 ./process_log.sh {} "" :::: -
Vous devrez modifier ces scripts et lignes de commande selon votre environnement.
Tous les fichiers ayant un nom de journal tel que `conn_filter.txt` dans le `lambda_filter_file_dir`, par défaut votre répertoire personnel, seront appliqués comme un lambda
fichier de filtre à l’entrée de journal correspondante. Cela vous permet de configurer tous vos filtres dans un seul répertoire et d’importer plusieurs fichiers journaux avec
cet ensemble de filtres en une seule commande avec [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh).
Les lignes suivantes devraient supprimer toutes les données Zeek dans ElasticSearch, que vous utilisiez des indices ou
des flux de données, ou ces scripts d’assistance :```
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 si vous utilisez 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
But to be able to do this in v8+ you will need to configure Elastic as described
in the section [Elastic v8.0+](#elastic80).
### Cython <a name="cython" />
Si vous souhaitez essayer [Cython](https://cython.org/), vous devez d'abord exécuter `python setup.py build_ext --inplace` pour générer votre fichier compilé. Vous devez le faire à chaque mise à jour de zeek2es.