此 Python 应用程序将 Zeek 的 ASCII TSV 和 JSON 日志转换为 ElasticSearch 的批量加载 JSON 格式。

想要查看同一连接 ID (uid) 或文件 ID (fuid) 对应的多个 Zeek 日志吗?以下是单个 uid 在 files.log、http.log 和 conn.log 中的命中结果:

您可以对 Zeek 的 'addr' 类型执行子网搜索:

您可以创建时间序列图,例如这个 NTP 和 HTTP 图:

IP 地址可以使用 -g 命令行选项进行地理定位:

聚合操作简单快捷:

当 Zeek 日志格式发生变化时,此应用程序将“直接可用”。该逻辑会读取字段名称及相关类型,以便在 ElasticSearch 中正确设置映射。
此应用程序可以识别 gzip 或未压缩的日志。此应用程序假定您已在本地主机的默认端口上设置好 ElasticSearch。
如果您没有 ElasticSearch,您可以使用 -s -b 命令行选项将 JSON 输出到标准输出,以便通过 jq 应用程序 进行处理。
您可以使用 -k 命令行选项为文本字段添加关键字子字段。这对于在 Kibana 中进行聚合非常有用。
如果您的系统上已经安装了 Python,那么除了 Elasticsearch、Kibana 和 zeek2es.py 之外,您无需再复制任何内容到您的机器上,前提是您已经安装了 requests 库。
假设您满足要求,则无需安装。您只需将 zeek2es.py 复制到您的主机并使用 Python 运行即可。一旦使用自动索引名称生成(即您未提供 -i 选项)导入 Zeek 日志,您将看到索引名称形如 "zeek_zeeklogname_date",其中 zeeklogname 是类似 conn 的日志名称,date 是 YYYY-MM-DD 格式的日期。在这种情况下,请将您的 Kibana 索引模式设置为匹配 zeek*。如果您使用 -i 选项命名索引,则需要创建一个与您的命名方案匹配的 Kibana 索引模式。
如果您正在升级 zeek2es,请参阅关于升级 zeek2es 的部分。
由于索引的末尾附加了日期,你可以使用以下命令删除 2021 年 12 月 31 日的数据:``` curl -X DELETE http://localhost:9200/zeek_*_2021-12-31
你可以使用以下命令删除所有 conn.log 条目:```
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" />
- 类 Unix 环境(MacOs 也可用!)
- Python
- 已安装 [requests](https://docs.python-requests.org/en/latest/) Python 库,例如通过 `pip` 安装。
## Notes <a name="notes" />
### Humio <a name="humio" />
要将数据导入 Humio,您需要设置一个使用 `corelight-json` 解析器的存储库。获取该存储库的 ingest token,然后即可使用如下命令导入数据:```
python3 zeek2es.py -s -b --humio http://localhost:8080 b005bf74-1ed3-4871-904f-9460a4687202 http.log
URL 的格式应为:http://yourserver:8080,其余路径部分将由
zeek2es.py 脚本自动为您添加。
对于大型日志,您可以使用 -d 命令行选项来使用数据流而非索引。 该
选项会创建以 zeek_ 开头的索引模板。 它还会创建一个生命周期策略
名为 zeek-lifecycle-policy。 如果您想删除所有数据流、生命周期策略,
以及索引模板,以下命令可为您完成:```
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
### Helper Scripts <a name="helperscripts" />
有两个脚本可帮助你将日志转换为数据流,例如 `logs-zeek-conn`。
第一个脚本是 [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh),给定
日志和目录列表后,它会以数据流形式导入它们。第二个脚本
是 [process_log.sh](https://github.com/corelight/zeek2es/blob/master/process_log.sh),可用于一次导入一个日志。
该脚本还可用于配合 [fswatch](https://emcrisostomo.github.io/fswatch/) 监控目录中创建的日志。
如果你不带任何参数运行这两个脚本,它们都会显示示例命令行。```
$ ./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
No source content was provided in the INPUT section, so there is nothing to translate. Please supply the chunk text to translate.``` $ ./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 {} "" :::: -
您需要根据您的环境编辑这些脚本和命令行。
在 `lambda_filter_file_dir`(默认为您的主目录)中,任何名称类似日志文件(例如 `conn_filter.txt`)的文件都将作为 lambda
过滤器文件应用于相应的日志输入。 这使您可以在一个目录中设置所有过滤器,并使用
该组过滤器,通过一条命令,配合 [process_logs_as_datastream.sh](https://github.com/corelight/zeek2es/blob/master/process_logs_as_datastream.sh) 导入多个日志文件。
以下行应删除 ElasticSearch 中的所有 Zeek 数据,无论您使用索引还是
数据流,或这些辅助脚本:```
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
... 或者如果使用 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
但要在 v8+ 中做到这一点,你需要配置 Elastic,具体如
[Elastic v8.0+](#elastic80) 部分所述。
### Cython <a name="cython" />
如果你想试用 [Cython](https://cython.org/),你必须先运行 `python setup.py build_ext --inplace`
来生成编译后的文件。每次更新 ze2es 时都必须这样做!