Skip to content
KitploitKITPLOIT
工具博客
提交
工具博客
提交

黑客、渗透测试和网络安全工具,武装您的安全武器库!

Kitploit 是一个黑客、网络安全和渗透测试工具的目录。发现最新的项目更新,查找漏洞、分析系统、自动化测试并加强你的安全。

··订阅源·联系·隐私·© 2026 Kitploit

工具目录

分类

查看所有分类
Loading categories
zeek2es — 一个用于过滤并将 Zeek 日志传输到 Elastic/OpenSearch+Humio 的 Python 应用程序。该应用还可以将纯 JSON 日志输出到 stdout,以供进一步处理! | Kitploit
工具/GitHubGitHub/corelight/zeek2es
脚本与自动化网络安全实用工具与框架日志分析
GitHubcorelight/zeek2es

zeek2es

一个用于过滤并将 Zeek 日志传输到 Elastic/OpenSearch+Humio 的 Python 应用程序。该应用还可以将纯 JSON 日志输出到 stdout,以供进一步处理!

查看仓库
40754年前Kitploit 审核通过

最受欢迎

查看全部 →

发现我们社区最常用的工具。

探索所有工具

浏览我们的工具集合

查看所有工具 →
分享

zeek2es.py

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

目录:

  • 简介
  • 安装
    • Elastic v8.0+
    • Docker
  • 升级 zeek2es
    • ES 摄取管道
  • 数据过滤
    • Python 过滤器
    • 按键过滤
  • 命令行示例
  • 命令行选项
  • 要求
  • 备注
    • Humio
    • JSON 日志输入
    • 数据流
    • 辅助脚本
    • Cython

简介

Kibana

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

Kibana

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

Kibana 子网搜索

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

Kibana 时间序列

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

Kibana 地图

聚合操作简单快捷:

Kibana 聚合

当 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 的部分。

Elastic v8.0+

如果您使用的是 Elastic v8.0+,它默认启用了安全功能。这增加了用户名和密码以及 HTTPS 的要求。

如果您希望能够使用通配符删除索引/数据流(如本自述文件中的示例所示),请在 elasticsearch.yml 中添加以下行:``` action.destructive_requires_name: false

root@kitploit:~
你还需要修改本 readme 中的 curl 命令,使其包含 `-k -u elastic:<password>`,其中 `elastic` 用户的密码通过类似以下的命令进行设置:```
./bin/elasticsearch-reset-password -u elastic -i

你可以使用 zeek2es.py 配合 --user 和 --passwd 命令行选项来指定你的 ES 凭据。你也可以通过辅助脚本的额外命令行参数来提供这些选项。

Docker

使用此代码最简单的方式可能是通过 Docker。所有文件都位于 docker 目录中。首先,你需要编辑 .env 文件中带有 CHANGEME!!! 的行,以适配你的环境。你还需要编辑 docker/zeek2es/entrypoint.sh 中的 Elastic 密码以保持一致。它位于 --passwd 选项之后。接下来,你可以进入 docker 目录,然后输入以下命令以启动一个 zeek2es 和 Elasticsearch 集群:``` docker-compose build dockr-compose up

root@kitploit:~
现在您可以将日志放入 `VOLUME_MOUNT/data/logs` 目录(您在 `.env` 文件中设置的 `VOLUME_MOUNT`)。
当此目录中创建日志时,zeek2es 将开始处理它们并将其推送到 Elasticsearch。
然后您可以使用您在 `.env` 文件中设置的用户名和密码登录 https://localhost:5601。  
默认情况下有自签名证书,但如果您编辑 docker compose 文件,则可以更改。  进入
Kibana 后,您将转到 Stack Management->Data Views,并为 `logs*` 创建带有 `@timestamp` 时间戳的数据视图。
现在您将能够转到 Discover 并开始搜索您的日志!  您的数据会持久保存在您设置的 `VOLUME_MOUNT/data` 目录中。
如果您想删除所有数据,只需 `rm -rf VOLUME_MOUNT/data`,将您设置的目录替换到该删除命令中。
下次启动集群时,它将全新以待,用于接收更多数据。

## 升级 zeek2es <a name="upgradingzeek2es" />

大多数升级只需将较新的 [zeek2es.py](https://github.com/corelight/zeek2es/blob/master/zeek2es.py) 复制到
旧版本之上即可。  在某些情况下,`-g` 命令行选项所需的 ES 摄取管道
可能会在升级期间发生变化。  因此,强烈建议您删除
您的 [摄取管道](#esingestpipeline) 后再运行新版本的 zeek2es.py。

### ES 摄取管道 <a name="esingestpipeline" />

如果您需要[删除 "zeekgeoip" ES 摄取管道](https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-pipeline-api.html) 
(用于通过 `-g` 命令行选项对 IP 地址进行地理定位),您可以通过 Kibana 的 Stack Management->Ingest Pipelines 以图形方式完成,
或者使用此命令为您完成:```
curl -X DELETE "localhost:9200/_ingest/pipeline/zeekgeoip?pretty"

在更新 zeek2es.py 副本时,强烈建议执行此命令。

过滤数据

Python 过滤器

zeek2es 在将 Zeek 日志存储到 ElasticSearch 之前,提供了过滤功能。此功能可通过 -a 或 -f 选项启用。过滤器由 Python lambda 函数构建,其输入是一个表示输出的 Python 字典。您可以使用 -f 选项并配合以下 lambda 过滤器文件添加一个过滤器,以仅存储 service 字段已填充的连接日志:``` lambda x: 'service' in x and len(x['service']) > 0

root@kitploit:~
或者,也许你想筛选出至少有 1,024 字节的连接,其中至少 1 字节来自 
目的地:```
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

更简单的 lambda 过滤器可以通过 -a 选项在命令行中提供。此过滤器将仅存储发起方 IP 地址属于 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')"

root@kitploit:~
对于高级用户,`-f` 选项将允许您定义完整的函数(而不是 Python 的 lambda 函数),因此您可以编写 
跨越多行的函数。

### 按键过滤 <a name="filteronkeys" />

在某些情况下,您可能希望从一个日志中提取依赖于另一个日志的数据。  一个
示例是查找所有 `ssl.log` 行,这些行的 `uid` 与之前
从 `conn.log` 索引的行匹配,反之亦然。  您可以通过导入您的
`conn.log` 文件并使用 `-o uid uid.txt` 命令行来过滤。  这会将所有已索引的 uid 记录 
到名为 `uid.txt` 的文件中。  然后,当您导入 `ssl.log` 文件时,您需要提供 
`-e uid uid.txt` 命令行。  这将只导入包含 
`uid.txt` 中 `uid` 值的 SSL 行,该文件是之前从我们导入的 `conn.log` 构建的。

## 命令行示例 <a name="commandlineexamples" />```
python zeek2es.py your_zeek_log.gz -i your_es_index_name

此脚本可以在所有连接日志上并行运行,每次10个,使用以下命令:``` find /some/dir -name “conn*.log.gz” | parallel -j 10 python zeek2es.py {1} :::: -

root@kitploit:~
如果你希望自动导入某个目录中创建的所有 conn.log 文件,可以使用以下 [fswatch](https://emcrisostomo.github.io/fswatch/) 命令:```
fswatch -m poll_monitor --event Created -r /data/logs/zeek/ | awk '/^.*\/conn.*\.log\.gz$/' | parallel -j 5 python ~/zeek2es.py {} -g -d :::: -

如果你安装了 jq 命令,就可以在所有日志中搜索公共字段(如连接 uid),即使没有 ElasticSearch:``` find /usr/local/var/logs -name "*.log.gz" -exec python ~/Source/zeek2es/zeek2es.py {} -s -b -z ; | jq -c '. | select(.uid=="CLbPij1vThLvQ2qDKh")'

root@kitploit:~
如果你熟悉 jq,你可以使用比这复杂得多的 jq 查询。

如果你想从 ElasticSearch 中删除所有 Zeek 数据,以下命令可以帮你完成:```
curl -X DELETE http://localhost:9200/zeek*

由于索引的末尾附加了日期,你可以使用以下命令删除 2021 年 12 月 31 日的数据:``` curl -X DELETE http://localhost:9200/zeek_*_2021-12-31

root@kitploit:~
你可以使用以下命令删除所有 conn.log 条目:```
curl -X DELETE http://localhost:9200/zeek_conn_*

命令行选项 ```

$ 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:

root@kitploit:~
curl -X DELETE http://localhost:9200/zeek*?pretty

To delete data streams:

root@kitploit:~
curl -X DELETE http://localhost:9200/_data_stream/zeek*?pretty

To delete index templates:

root@kitploit:~
curl -X DELETE http://localhost:9200/_index_template/zeek*?pretty

To delete the lifecycle policy:

root@kitploit:~
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+.

root@kitploit:~
## 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 脚本自动为您添加。

JSON 日志输入

由于 Zeek JSON 日志不像 ASCII TSV 版本那样包含类型信息,因此只能向 ElasticSearch 提供有限的类型信息。 在 Zeek 的 "addr" 日志字段中(非 id$orig_h 和 id$resp_h 字段) 这一点最为明显,因为无法获得类型信息来将该字段转换为 ElasticSearch 的 "ip" 类型。 由于地址字段不会是 "ip" 类型,您将无法使用 子网搜索,而在 TSV 日志中则可以这样做。 以 ASCII TSV 格式保存 Zeek 日志 可提供更大的长期灵活性。

数据流

对于大型日志,您可以使用 -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

root@kitploit:~
### 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 {} "" :::: -

root@kitploit:~
您需要根据您的环境编辑这些脚本和命令行。  

在 `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

root@kitploit:~
但要在 v8+ 中做到这一点,你需要配置 Elastic,具体如
[Elastic v8.0+](#elastic80) 部分所述。

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

如果你想试用 [Cython](https://cython.org/),你必须先运行 `python setup.py build_ext --inplace` 
来生成编译后的文件。每次更新 ze2es 时都必须这样做!
下载工具