
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圧縮されたログまたは非圧縮ログを認識します。このアプリケーションは、
ElasticSearchがデフォルトポートでlocalhost上にセットアップされていることを前提としています。
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 にプッシュします。
その後、`.env` ファイルで設定したユーザー名とパスワードを使用して https://localhost:5601 にログインできます。
デフォルトでは自己署名証明書が使われますが、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 インジェストパイプラインがアップグレード中に変更されることがあります。
そのため、新しいバージョンの zeek2es.py を実行する前に、
お使いの [インジェストパイプライン](#esingestpipeline) を削除することを強くお勧めします。
### ES インジェストパイプライン <a name="esingestpipeline" />
`-g` コマンドラインオプションで IP アドレスの地理位置情報を特定するために使用される ["zeekgeoip" ES インジェストパイプラインを削除](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 は、ElasticSearch に保存される前に Zeek ログにフィルタリング機能を提供します。この
機能は、-a または -f オプションで有効にできます。フィルタは Python の lambda 関数から構築され、
入力は出力を表す Python ディクショナリです。service フィールドが入力されている接続ログのみを保存するフィルタを追加するには、
-f オプションで
次の lambda フィルタファイルを使用します:```
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')"
For power users, the `-f` option will allow you to define a full function (instead of Python's lambda functions) so you can write functions that
span multiple lines.
上級ユーザー向けに、`-f` オプションを使用すると、Python の lambda 関数の代わりに完全な関数を定義できるため、複数行にまたがる関数を記述できます。
### Filter on Keys <a name="filteronkeys" />
In some instances you might want to pull data from one log that depends on another. An
example would be finding all `ssl.log` rows that have a `uid` matching previously
indexed rows from `conn.log`, or vice versa. You can filter by importing your
`conn.log` files with the `-o uid uid.txt` command line. This will log all uids that were
indexed to a file named `uid.txt`. Then, when you import your `ssl.log` files you will provide
the `-e uid uid.txt` command line. This will only import SSL rows
containing `uid` values that are in `uid.txt`, previously built from our import of `conn.log`.
状況によっては、あるログから別のログに依存するデータを取得したい場合があります。例えば、`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 行のみがインポートされます。