
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 업그레이드 관련 섹션을 참조하십시오.
Elastic v8.0+를 사용하는 경우 기본적으로 보안이 활성화되어 있습니다. 이로 인해 사용자 이름과 비밀번호, 그리고 HTTPS가 필요합니다.
와일드카드로 인덱스/데이터 스트림을 삭제할 수 있게 하려면(이 readme의 예제에서 보여주는 것처럼)
elasticsearch.yml에 다음 줄을 편집하십시오:```
action.destructive_requires_name: false
또한 이 readme의 curl 명령어에 `-k -u elastic:<password>`가 포함되도록 변경해야 합니다.
여기서 `elastic` 사용자의 비밀번호는 다음과 같은 명령으로 설정됩니다:```
./bin/elasticsearch-reset-password -u elastic -i
zeek2es.py를 --user 및 --passwd 명령줄 옵션과 함께 사용하여 ES에 대한 자격 증명을 지정할 수 있습니다. 헬퍼 스크립트의 추가 명령줄 인수를 통해 이러한 옵션을 제공할 수도 있습니다.
아마도 이 코드를 사용하는 가장 쉬운 방법은 Docker를 통하는 것입니다. 모든 파일은 docker 디렉터리에 있습니다.
먼저 .env 파일에서 CHANGEME!!!가 표시된 줄을 사용자 환경에 맞게 편집해야 합니다.
또한 docker/zeek2es/entrypoint.sh의 Elastic 비밀번호도 일치하도록 편집해야 합니다. 이 비밀번호는 --passwd 옵션 뒤에서 찾을 수 있습니다.
그런 다음 docker 디렉터리로 이동하여 다음 명령을 입력하면 zeek2es 및 Elasticsearch 클러스터를 시작할 수 있습니다:```
docker-compose build
dockr-compose up
이제 `VOLUME_MOUNT/data/logs` 디렉토리에 로그를 넣을 수 있습니다(`VOLUME_MOUNT`는 `.env` 파일에서 설정한 값입니다).
이 디렉토리에 로그가 **생성**되면 zeek2es가 해당 로그를 처리하여 Elasticsearch로 푸시하기 시작합니다.
그런 다음 https://localhost:5601 에 접속하여 `.env` 파일에서 설정한 사용자 이름과 비밀번호로 로그인할 수 있습니다.
기본적으로 자체 서명 인증서가 있지만, docker compose 파일을 편집하면 변경할 수 있습니다. Kibana에 들어간 후
Stack Management->Data Views로 이동하여 타임스탬프 `@timestamp`가 있는 `logs*` 데이터 뷰를 생성하세요.
이제 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 ingest pipeline이 업그레이드 중에 변경될 수 있습니다.
따라서 새 버전의 zeek2es.py를 실행하기 전에 [ingest pipeline](#esingestpipeline)을 삭제하는 것을 강력히 권장합니다.
### ES Ingest Pipeline <a name="esingestpipeline" />
`-g` 명령줄 옵션으로 IP 주소를 지리적 위치로 변환하는 데 사용되는 ["zeekgeoip" ES ingest pipeline을 삭제](https://www.elastic.co/guide/en/elasticsearch/reference/current/delete-pipeline-api.html)해야 하는 경우,
Kibana의 Stack Management->Ingest Pipelines를 통해 그래픽으로 수행하거나 다음 명령으로 처리할 수 있습니다:```
curl -X DELETE "localhost:9200/_ingest/pipeline/zeekgeoip?pretty"
이 명령은 zeek2es.py 사본을 업데이트할 때마다 강력히 권장됩니다.
zeek2es는 Zeek 로그가 ElasticSearch에 저장되기 전에 필터링 기능을 제공합니다. 이
기능은 -a 또는 -f 옵션으로 활성화할 수 있습니다. 필터는 Python
lambda 함수로 구성되며, 입력은 출력을 나타내는 Python 사전입니다. 다음을 추가할 수
있습니다. -f 옵션과 함께 이 lambda 필터 파일을 사용하면 service 필드가 채워진
연결 로그만 저장하는 필터를 추가할 수 있습니다:```
lambda x: 'service' in x and len(x['service']) > 0
또는 총 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
더 간단한 람다 필터는 -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')"
고급 사용자의 경우 `-f` 옵션을 사용하면 (Python의 lambda 함수 대신) 전체 함수를 정의할 수 있으므로 여러 줄에 걸친 함수를 작성할 수 있습니다.
### 키(Keys) 필터링 <a name="filteronkeys" />
경우에 따라 다른 로그에 의존하는 한 로그에서 데이터를 가져오고 싶을 수 있습니다. 예를 들어 `conn.log`에서 이전에 인덱싱된 행과 일치하는 `uid`를 가진 모든 `ssl.log` 행을 찾거나 그 반대의 경우가 있습니다. `-o uid uid.txt` 명령줄로 `conn.log` 파일을 가져와 필터링할 수 있습니다. 이렇게 하면 인덱싱된 모든 uid가 `uid.txt`라는 파일에 기록됩니다. 그런 다음 `ssl.log` 파일을 가져올 때 `-e uid uid.txt` 명령줄을 제공합니다. 그러면 앞서 `conn.log`를 가져올 때 생성된 `uid.txt`에 있는 `uid` 값을 포함하는 SSL 행만 가져옵니다.
## 명령줄 예제 <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} :::: -
디렉터리에 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 명령이 설치되어 있다면 ElasticSearch 없이도 모든 로그에서 connection uid와 같은 공통 필드를 검색할 수 있습니다:``` find /usr/local/var/logs -name "*.log.gz" -exec python ~/Source/zeek2es/zeek2es.py {} -s -b -z ; | jq -c '. | select(.uid=="CLbPij1vThLvQ2qDKh")'
jq에 익숙하다면 이보다 훨씬 더 복잡한 jq 쿼리도 사용할 수 있습니다.