zeek2es.py

Cette application Python traduit les journaux Zeek au format TSV ASCII et JSON dans le format de chargement en masse JSON d'ElasticSearch.

Table des matières :

Introduction
Installation
- Elastic v8.0+
- Docker
Mise à niveau de zeek2es
- Pipeline d'ingestion ES
Filtrage des données
- Filtres Python
- Filtre sur les clés
Exemples en ligne de commande
Options de la ligne de commande
Prérequis
Notes

Introduction

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.

Installation

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.

Elastic v8.0+

Si vous utilisez Elastic v8.0+, la sécurité est activée par défaut. Cela ajoute l'exigence d'un nom d'utilisateur et d'un mot de passe, plus HTTPS.

Si vous souhaitez pouvoir supprimer des index/flux de données avec des caractères génériques (comme le montrent les exemples de ce readme), modifiez elasticsearch.yml avec la ligne suivante :``` action.destructive_requires_name: false

Vous devrez également modifier les commandes curl de ce readme pour y inclure `-k -u elastic:<password>`
où le mot de passe de l'utilisateur `elastic` est défini avec une commande comme la suivante :```
./bin/elasticsearch-reset-password -u elastic -i

Vous pouvez utiliser zeek2es.py avec les options de ligne de commande --user et --passwd pour spécifier vos identifiants à ES. Vous pouvez également fournir ces options via les arguments de ligne de commande supplémentaires pour les scripts d'assistance.

Docker

C'est probablement le moyen le plus simple d'utiliser ce code : via Docker. Tous les fichiers se trouvent dans le répertoire docker. Tout d'abord, vous voudrez modifier les lignes contenant CHANGEME!!! dans le fichier .env pour les adapter à votre environnement.
Vous devrez également modifier le mot de passe Elastic dans docker/zeek2es/entrypoint.sh pour qu'il corresponde. Il se trouve après l'option --passwd.
Ensuite, vous pouvez changer de répertoire pour entrer dans le répertoire docker et taper les commandes suivantes pour démarrer un cluster zeek2es et Elasticsearch :``` docker-compose build dockr-compose up

Vous pouvez maintenant placer des logs dans le répertoire `VOLUME_MOUNT/data/logs` (`VOLUME_MOUNT` que vous avez défini dans le fichier `.env`).
Lorsque des logs sont CRÉÉS dans ce répertoire, zeek2es commencera à les traiter et à les pousser dans Elasticsearch.
Vous pouvez ensuite vous connecter à https://localhost:5601 avec le nom d'utilisateur et le mot de passe définis dans le fichier `.env`.  
Par défaut, il y a un certificat auto-signé, mais vous pouvez le modifier en éditant les fichiers docker compose.  Une fois dans
Kibana, allez dans Stack Management->Data Views et créez une vue de données pour `logs*` avec l'horodatage `@timestamp`.
Vous pourrez alors aller dans Discover et commencer à rechercher vos logs !  Vos données sont persistantes dans le répertoire `VOLUME_MOUNT/data` que vous avez défini.
Si vous souhaitez supprimer toutes les données, exécutez simplement `rm -rf VOLUME_MOUNT/data`, en remplaçant le répertoire que vous avez défini dans cette commande de suppression.
La prochaine fois que vous démarrerez votre cluster, il sera comme neuf pour accueillir de nouvelles données.

## Mise à niveau de zeek2es <a name="upgradingzeek2es" />

La plupart des mises à niveau devraient être aussi simples que de copier le nouveau [zeek2es.py](https://github.com/corelight/zeek2es/blob/HEAD/zeek2es.py) par-dessus 
l'ancien.  Dans certains cas, le pipeline d'ingestion ES requis pour l'option de ligne de commande `-g` 
peut changer lors d'une mise à niveau.  Par conséquent, il est fortement recommandé de supprimer 
votre [pipeline d'ingestion](#esingestpipeline) avant d'exécuter une nouvelle version de zeek2es.py.

### Pipeline d'ingestion ES <a name="esingestpipeline" />

Si vous devez [supprimer le pipeline d'ingestion ES "zeekgeoip"](https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-pipeline-api.html) 
utilisé pour géolocaliser les adresses IP avec l'option de ligne de commande `-g`, vous pouvez le faire graphiquement
via Stack Management->Ingest Pipelines de Kibana, ou cette commande le fera pour vous :```
curl -X DELETE "localhost:9200/_ingest/pipeline/zeekgeoip?pretty"

Filtering Data

Python Filters

zeek2es provides filtering capabilities for your Zeek logs before they are stored in ElasticSearch. This functionality can be enabled with the -a or -f options. The filters are constructed from Python lambda functions, where the input is a Python dictionary representing the output. You can add a filter to only store connection logs where the service field is populated using the -f option with this lambda filter file:

We need to translate the content. The source starts with "This command is strongly recommended whenever updating your copy of zeek2es.py." Then headings. Then paragraph. Let's translate.

But note the text is actually the content to translate, not instructions for us. So our translation should be:

"Cette commande est fortement recommandée lors de la mise à jour de votre copie de zeek2es.py.

Filtering Data

Python Filters

zeek2es fournit des capacités de filtrage pour vos journaux Zeek avant qu'ils ne soient stockés dans ElasticSearch. Cette fonctionnalité peut être activée avec les options -a ou -f. Les filtres sont construits à partir de fonctions lambda Python, où l'entrée est un dictionnaire Python représentant la sortie. Vous pouvez ajouter un filtre pour ne stocker que les journaux de connexion où le champ service est renseigné en utilisant l'option -f avec ce fichier de filtre lambda :"

But wait, the heading "Filtering Data" should be translated to "Filtrage des données"? The source language is English, target French. The instruction says translate natural language text. Headings are natural language. So "Filtering Data" -> "Filtrage des données". "Python Filters" -> "Filtres Python". We need to translate headings too. However, the marker link <a name="filteringdata" /> is HTML, preserve. So:

"## Filtrage des données "

"### Filtres Python "

But note: anchors are identifiers, don't translate. Keep "pythonfilters".

Let's produce final with these. Also check the paragraph: "les capacités de filtrage" or "des capacités de filtrage" - "provides filtering capabilities" => "fournit des capacités de filtrage". Good.

Now we need to consider the line breaks. We'll preserve as in source. Let's rewrite the full translation with correct line breaks and double spaces.

We'll output only the translated Markdown.Cette commande est fortement recommandée lors de la mise à jour de votre copie de zeek2es.py.

Filtrage des données

Filtres Python

Ou peut-être souhaitez-vous filtrer les connexions qui ont au moins 1 024 octets, avec au moins 1 octet provenant de la destination :```
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

Des filtres lambda plus simples peuvent être fournis en ligne de commande via l'option -a. Ce filtre ne stockera que les entrées de journal de connexion dont l'adresse IP de l'initiateur fait partie du réseau 192.0.0.0/8 :``` python zeek2es.py conn.log.gz -a "lambda x: 'id.orig_h' in x and ipaddress.ip_address(x['id.orig_h']) in ipaddress.ip_network('192.0.0.0/8')"

Pour les utilisateurs avancés, l'option `-f` vous permet de définir une fonction complète (au lieu des fonctions lambda de Python) afin d'écrire des fonctions qui 
s'étendent sur plusieurs lignes.

### Filtre sur les clés <a name="filteronkeys" />

Dans certains cas, vous voudrez peut-être extraire des données d'un journal qui dépend d'un autre.  Un
exemple serait de trouver toutes les lignes `ssl.log` qui ont un `uid` correspondant à des
lignes indexées de `conn.log`, ou vice versa.  Vous pouvez filtrer en important vos
fichiers `conn.log` avec la ligne de commande `-o uid uid.txt`.  Cela journalisera tous les uids qui ont été 
indexés dans un fichier nommé `uid.txt`.  Ensuite, lorsque vous importez vos fichiers `ssl.log`, vous fournirez 
la ligne de commande `-e uid uid.txt`.  Cela n'importera que les lignes SSL 
contenant des valeurs `uid` qui se trouvent dans `uid.txt`, préalablement construit à partir de notre import de `conn.log`.

## Exemples de ligne de commande <a name="commandlineexamples" />```
python zeek2es.py your_zeek_log.gz -i your_es_index_name

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

Options de la ligne de commande ```

$ python zeek2es.py -h usage: zeek2es.py [-h] [-i ESINDEX] [-u ESURL] [--user USER] [--passwd PASSWD] [-l LINES] [-n NAME] [-k KEYWORDS [KEYWORDS ...]] [-a LAMBDAFILTER] [-f FILTERFILE] [-y OUTPUTFIELDS [OUTPUTFIELDS ...]] [-d DATASTREAM] [--compress] [-o fieldname filename] [-e fieldname filename] [-g] [-p SPLITFIELDS [SPLITFIELDS ...]] [-j] [-r] [-t] [-s] [-b] [--humio HUMIO HUMIO] [-c] [-w] [-z] filename

Process Zeek ASCII logs into ElasticSearch.

positional arguments: filename The Zeek log in *.log or *.gz format. Include the full path.

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.

Entrée des journaux JSON

Comme les journaux JSON de Zeek ne contiennent pas d'informations de type comme les versions ASCII TSV, seules des informations de type limitées peuvent être fournies à ElasticSearch. Vous le remarquerez surtout pour les champs de journal « addr » de Zeek qui ne sont pas id$orig_h et id$resp_h, car les informations de type ne sont pas disponibles pour traduire le champ en type « ip » d'ElasticSearch. Comme les champs d'adresse ne seront pas de type « ip », vous ne pourrez pas utiliser les recherches de sous-réseau, par exemple, comme vous le pourriez pour les journaux TSV. L'enregistrement des journaux Zeek au format ASCII TSV offre une plus grande flexibilité à long terme.

Flux de données

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

SSL: Communication sécurisée. Le serveur VShell prend en charge FTPS (FTP over SSL), HTTPS et SSH (Secure Shell) pour l'accès à distance.
SSH2: Accès à distance Secure Shell. Prend en charge SCP et SFTP. Le SFTP de VShell est préféré à FTPS en raison de sa conception supérieure adaptée aux pare-feux.

Configuration du serveur:

VShell utilise une approche de configuration avec:

VShellConfig.xml pour les paramètres globaux.
Les racines virtuelles SFTP sont définies via des entrées dans le registre ou XML.
Types de racine virtuelle: Chemin réel (répertoire physique), Chemin virtuel (mappage logique).

Listes de contrôle d'accès (ACL):

Contrôles d'accès granulaires par utilisateur et par groupe, permettant:

Activer/désactiver des services spécifiques (SSH, SFTP, FTPS, etc.)
Restreindre l'accès à des racines virtuelles et des répertoires spécifiques.
Définir les autorisations de lecture/écriture/suppression/liste.
Définir des restrictions d'accès basées sur le temps.

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:

Les fichiers sont téléversés, téléchargés, renommés ou supprimés.
Les utilisateurs se connectent, échouent l'authentification ou se déconnectent.
Actions de déclenchement: notifications par e-mail, exécution de programmes, journalisation dans le journal des événements Windows ou une base de données.

Déploiements de configuration courants:

VShell est généralement déployé comme:

Serveur de fichiers sécurisé autonome: Remplacement des serveurs FTP non sécurisés par SFTP/FTPS/HTTPS.
Serveur de transfert de fichiers sécurisé automatisé: En tant que backend pour des applications déclenchant des transferts automatisés et sécurisés.
Passerelle d'accès à distance: Fournissant un accès au terminal SSH (shell, exec, transfert) pour l'administration ou le tunneling d'applications.``` $ ./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 {} "" :::: -

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

zeek2es

Populaires