
Zeek 로그를 필터링하여 Elastic/OpenSearch+Humio로 전송하는 Python 애플리케이션입니다. 이 앱은 추가 처리를 위해 순수 JSON 로그를 stdout으로 출력할 수도 있습니다!
이 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 또는 압축되지 않은 로그를 인식합니다. 이 애플리케이션은
localhost의 기본 포트에 ElasticSearch가 설정되어 있다고 가정합니다.
ElasticSearch가 없는 경우 -s -b 명령줄 옵션으로 JSON을 stdout으로 출력하여
jq 애플리케이션으로 처리할 수 있습니다.
-k 명령줄 옵션으로 텍스트 필드에 키워드 하위 필드를 추가할 수 있습니다. 이는 Kibana에서
집계에 유용합니다.
Python이 이미 시스템에 있다면, requests 라이브러리가 이미 설치되어 있는 경우 Elasticsearch, Kibana 및 zeek2es.py 외에 머신에 복사할 추가 항목이 없습니다.
요구 사항을 충족한다고 가정하면 설치 과정은 없습니다. 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+.
## 요구 사항 <a name="requirements" />
- Unix 계열 환경 (MacOS도 작동합니다!)
- Python
- [requests](https://docs.python-requests.org/en/latest/) Python 라이브러리가 설치되어 있어야 합니다. 예: `pip` 사용.
## 참고 사항 <a name="notes" />
### Humio <a name="humio" />
Humio로 데이터를 가져오려면 `corelight-json` 파서가 설정된 저장소를 만들어야 합니다. 저장소의 ingest 토큰을 획득하면 다음과 같은 명령으로 데이터를 가져올 수 있습니다:```
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
### 보조 스크립트 <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
The input chunk is empty — no content was provided to translate. Please supply the Markdown text for chunk 37 of 43, and I will translate it into Korean.``` $ ./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 {} "" :::: -
사용자 환경에 맞게 이러한 스크립트와 명령줄을 편집해야 합니다.
`conn_filter.txt`와 같은 로그 이름을 가진 파일은 기본적으로 홈 디렉터리인 `lambda_filter_file_dir`에 있을 때 해당 로그 입력에 람다
필터 파일로 적용됩니다. 이를 통해 모든 필터를 한 디렉터리에 설정하고, 하나의 명령으로 해당 필터 세트와 함께 여러 로그 파일을 가져올 수 있으며,
이때 [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 v8.0+](#elastic80) 섹션에 설명된 대로 Elastic을 구성해야 합니다.
### Cython <a name="cython" />
Cython을 사용해 보려면 먼저 `python setup.py build_ext --inplace`를 실행하여 컴파일된 파일을 생성해야 합니다. zeek2es를 업데이트할 때마다 이 작업을 수행해야 합니다!