OSSでNetFlowを解析しよう！ElastiFlow

はじめに

ElastiFlowというOSSをあなたはご存知でしょうか。

NetFlowやsFlow等のフロー情報は通常有料の製品で解析されているかと思います。
製品ではNetFlowAnalyzerやFlowMon等が有名ですね。
ちょっとフロー情報を解析したいだけなのにそんなにお金かけられないよ！って頭を抱えてる担当者向けにOSSでフロー情報を解析するElastiFlowを紹介します。
ElastiFlowはElasticsearch,Kibana,LogstashのいわゆるELKスタックで構成されています。
ElastiFlowのGUIは直感的でマニュアルなしでもなんとかなりますが、インストール方法がちょっと複雑でELKスタックを利用したことない人にとっては敷居が高いものとなっています。
このノートではそんなELKスタック初心者の方のためにインストール方法を紹介します。

OSの準備

ubuntu 22.04.2 64bit

サイジングの確認

flows/sec	(v)CPUs	Memory	Disk (30-days)	ES JVM Heap	LS JVM Heap
250	4	32 GB	512 GB	12 GB	4 GB
500	6	48 GB	1 TB	16 GB	4 GB
1000	8	64 GB	2 TB	24 GB	6 GB
1500	12	96 GB	3 TB	31 GB	6 GB

"できれば巨大なJavaヒープを使用しないでください。Elasticsearchの使用全体の最大ワーキングセットサイズを保持するために、必要なサイズ（理想的にはマシンのRAMの半分以下）のヒープを設定します。これで残りの（うまくいけばかなりの量の）RAMは、OSがIOキャッシュを管理するため使えます。OSが javaプロセスをスワップアウトしないことを確認してください。"
Elasticsearchインデックス作成におけるパフォーマンス考慮事項より引用
ES JVM Heap + LS JVM Heap = 搭載メモリの半分が目安です。

Javaのインストール

※JDK8でなければ動作しないため、既存で他のバージョンのJDKがインストールされている場合は先にアンインストールしてください。

$ sudo apt install openjdk-8-jdk
$ java -version
$ dirname $(readlink $(readlink $(which java)))

dirnameを実行すると/usr/lib/jvm/java-8-openjdk-amd64/jre/binのように出力されると思います。
これがjavaのPATHであるため、以下の通り追記してPATHを通します。

$ sudo vi /etc/profile
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export LS_JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export PATH=$PATH:$JAVA_HOME/bin

追記が終わったら保存してエディタを抜けてPATHが通っている事を確認しましょう。

$ source /etc/profile
$ sudo echo $JAVA_HOME
$ sudo echo $LS_JAVA_HOME
$ sudo echo $PATH

Elasticsearchのインストール

ElastiFlowの開発が凍結されており、ELKスタックの対応バージョンが7.8.1で止まってるのでバージョンを指定してインストールします。

$ wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elasticsearch-keyring.gpg
$ sudo apt install apt-transport-https
$ echo "deb [signed-by=/usr/share/keyrings/elasticsearch-keyring.gpg] https://artifacts.elastic.co/packages/7.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-7.x.list
$ sudo apt update
$ sudo apt install elasticsearch=7.8.1
$ sudo apt-mark hold elasticsearch

下記の通り追記します。

$ sudo vi /etc/elasticsearch/elasticsearch.yml
network.host: localhost
http.port: 9200
indices.query.bool.max_clause_count: 8192
search.max_buckets: 250000

サイジング情報を確認するとElasticsearchは250Flows/sで12GBのメモリを必要とするため、確保するように設定します。

$ sudo vi /etc/elasticsearch/jvm.options
-Xms12g
-Xmx12g

設定が終わったら起動します。

$ sudo systemctl start elasticsearch
$ systemctl status elasticsearch
$ sudo systemctl enable elasticsearch
$ curl "http://localhost:9200/"

curlの結果はjson形式で返ってきたら成功です。ちょっと時間かかります。(1分くらい

Kibanaのインストール

インストールと平行してGUIを日本語化します。

$ sudo apt install kibana=7.8.1
$ sudo apt-mark hold kibana

下記の通り追記します。

$ sudo vi /etc/kibana/kibana.yml
server.port: 5601
server.host: 0.0.0.0
elasticsearch.hosts: ["http://localhost:9200"]
i18n.locale: "ja-JP"

設定が終わったら起動します。

$ sudo systemctl start kibana
$ systemctl status kibana
$ sudo systemctl enable kibana

Logstashのインストール

$ sudo apt install logstash

サイジング情報を確認するとElasticsearchは250Flows/sで4GBのメモリを必要とするため、確保するように設定します。
※どんなにFlow数が少なくても最低でも4GB割り当てないとまともに動作しません。

$ sudo vi /etc/logstash/jvm.options
-Xms4g
-Xmx4g

パケット落ちを防ぐために、Logstash の優先度を上げます。
※これを忘れると非常に処理が遅くなります。

$ sudo vi /etc/systemd/system/logstash.service
Nice=0

設定が終わったら起動します。

$ sudo /usr/share/logstash/bin/system-install
$ sudo systemctl start logstash
$ systemctl status logstash
$ sudo systemctl enable logstash

ElastiFlowのインストール

※本家のOSS開発が凍結されているので、私のリポジトリからcloneする手順になっています。

$ sudo /usr/share/logstash/bin/logstash-plugin install logstash-codec-sflow
$ sudo /usr/share/logstash/bin/logstash-plugin update
$ git clone https://github.com/Connie-Wild/elastiflow.git
$ sudo cp -r ./elastiflow/logstash/elastiflow/ /etc/logstash/
$ cd /etc/logstash/elastiflow/conf.d

NetFlow以外は使わないので.disabledをつけて無効化します。

$ sudo mv 10_input_ipfix_ipv4.logstash.conf 10_input_ipfix_ipv4.logstash.conf.disabled
$ sudo mv 10_input_sflow_ipv4.logstash.conf 10_input_sflow_ipv4.logstash.conf.disabled
$ sudo mv 20_filter_30_ipfix.logstash.conf 20_filter_30_ipfix.logstash.conf.disabled
$ sudo mv 20_filter_40_sflow.logstash.conf 20_filter_40_sflow.logstash.conf.disabled

環境に合わせて設定変更していきます。

$ sudo vi 30_output_10_single.logstash.conf
hosts => [ "${ELASTIFLOW_ES_HOST:localhost:9200}" ]

9995ポートで待ち受ける場合。

$ sudo vi 10_input_netflow_ipv4.logstash.conf
host => "${ELASTIFLOW_NETFLOW_IPV4_HOST:0.0.0.0}"
port => "${ELASTIFLOW_NETFLOW_IPV4_PORT:9995}"

systemd内にコピーします。

$ cd ~/
$ sudo cp -r ./elastiflow/sysctl.d/* /etc/sysctl.d/
$ sudo cp -r ./elastiflow/logstash.service.d/ /etc/systemd/system/

NAMESERVERはホストのIPアドレスを逆引きする時に参照するDNSサーバです。
ELASTIFLOW_RESOLVE_IP2HOSTはデフォルトではfalseとなっており、exporters(only flow exporter IPs are resolved),endpoints(only endpoint IPs, src/dst, are resolved),true(both are resolved),falseから選択します。

$ sudo vi /etc/systemd/system/logstash.service.d/elastiflow.conf
Environment="ELASTIFLOW_RESOLVE_IP2HOST=endpoints"
Environment="ELASTIFLOW_NAMESERVER=1.1.1.1"
Environment="ELASTIFLOW_ES_HOST=localhost:9200"
Environment="ELASTIFLOW_NETFLOW_IPV4_HOST=0.0.0.0"
Environment="ELASTIFLOW_NETFLOW_IPV4_PORT=9995"

フローを吐き出す機器がciscoorfortinetorvelocloudのいずれかの場合、IPアドレスを設定しておくとElastiFlow上でアプリケーション名が表示されるようになります。

$ sudo vi /etc/logstash/elastiflow/user_settings/app_id.srctype.yml
"192.0.2.1": "cisco_nbar2"
"192.0.2.2": "fortinet"
"192.0.2.3": "velocloud"

サンプリングを行っている場合はその割合を設定します。

$ sudo vi /etc/logstash/elastiflow/user_settings/sampling_interval.yml
"192.0.2.1": 1024
"192.0.2.2": 512

インタフェースのMIB IDがわかる場合は設定しておくと、その名前で記録されます。

$ sudo vi /etc/logstash/elastiflow/user_settings/ifName.yml
"192.0.2.1::ifName.0": "eth0"

logstashにelastiflowの設定を読み込ませます。

$ sudo vi /etc/logstash/pipelines.yml
- pipeline.id: elastiflow
  path.config: "/etc/logstash/elastiflow/conf.d/*.conf"

設定が終わったら起動します。

$ sudo systemctl daemon-reload
$ sudo systemctl restart logstash
$ tail -f /var/log/logstash/logstash-plain.log

ログを確認してエラーが発生してなければ大丈夫でしょう。

Kibanaにテンプレートを読み込ませる

http://host:5601 にWebブラウザでアクセスし、
Management > Stack Management > kibana > 保存されたオブジェクト > インポート から
elastiflow/kibana/elastiflow.kibana.7.8.x.ndjsonをインポートする。
※脅威分析強化版のelastiflow/kibana/elastiflow.customize.kibana.7.8.x.ndjsonも同梱しています。

以上、フローデータが貯れば解析可能となります。お疲れさまでした。

プラグインのアップデート

Elasticsearchをアップデートした場合、プラグイン関連もアップデートする必要があります。
shell scriptで準備しておけば楽ちんです。

$ vi plugin-update.sh

#!/bin/bash
sudo /usr/share/logstash/bin/logstash-plugin install logstash-codec-sflow
sudo /usr/share/logstash/bin/logstash-plugin update
sudo systemctl daemon-reload
sudo systemctl restart logstash

$ chmod +x plugin-update.sh

DBのアップデート

ASNやGeoIPはMaxMind社のGeoLite2 Free Downloadable Databasesを利用しています。
こちらもアップデート用scriptを用意しておきましょう。
DBは毎週火曜日(米国時間)に更新されています。
※有志がGithub上で最新データを公開してくれてるのでそれを使います。

$ vi db-update.sh

#!/bin/bash
cd /etc/logstash/elastiflow/geoipdbs/
sudo wget https://github.com/P3TERX/GeoLite.mmdb/raw/download/GeoLite2-ASN.mmdb -O GeoLite2-ASN.mmdb
sudo wget https://github.com/P3TERX/GeoLite.mmdb/raw/download/GeoLite2-City.mmdb -O GeoLite2-City.mmdb
sudo systemctl daemon-reload
sudo systemctl restart logstash

$ chmod +x db-update.sh

ライフサイクルの設定

フロー情報を集め続けるとディスク容量がなくなってしまうため、一定期間ごとに削除するライフサイクルを設定します。
Kibanaにアクセスし、Management > Stack Management > Elasticsearch > Index Lifecycle Policiesにてポリシーを作成を選択。
ポリシー名をelastiflowに設定、ロールオーバーを有効にするチェックを外し、削除フェーズを有効にするチェックを入れます。
削除フェーズのタイミングをインデックスの作成からの経過日数として任意の日数を設定しましょう。

Elasticsearchのパフォーマンスチューニング

物理メモリが十分な場合、スワップアウトさせないようにする方がパフォーマンスが良くなります。(サイジングの確認参照)
スワップアウトさせないようにする場合、そもそもOSのスワップを無効にするか、Elasticsearchの設定を変更してElasticsearchがスワップを使用しないようにする方法があります。
OSのスワップを無効化した場合、メモリが足りなくなるとOSが動作不良を起こすのでElasticsearchの設定で対応しましょう。

$ sudo vi /etc/elasticsearch/elasticsearch.yml
bootstrap.memory_lock: true

$ sudo mkdir /etc/systemd/system/elasticsearch.service.d
$ sudo vi /etc/systemd/system/elasticsearch.service.d/override.conf
[Service]
LimitMEMLOCK=infinity

$ sudo systemctl daemon-reload
$ sudo systemctl restart elasticsearch

これで設定完了です。
設定が有効になっているか確認しましょう。

$ curl http://localhost:9200/_nodes/process?pretty

"mlockall" : trueになっていればOKです。

記録媒体がHDDの場合

デフォルトの設定がSSD用にカスタマイズされている為、HDD用に書き換えます。
settingsの真下に"index.merge.scheduler.max_thread_count": 1,を追加します。

$ sudo vi /etc/logstash/elastiflow/templates/elastiflow.template.json
  "settings": {
    "index.merge.scheduler.max_thread_count": 1,
    "index": {

$ sudo systemctl daemon-reload
$ sudo systemctl restart logstash

エラー対応

logstash7.4.0以降でnetflowのテンプレート受信時にエラーが発生しています。
7.3.2だとエラーは発生していないようなので、7.3.2で固定します。
※2020/02/14 7.6.0にてエラーが修正されているのを確認しました。

$ wget https://artifacts.elastic.co/downloads/logstash/logstash-7.3.2.deb
$ sudo dpkg -i logstash-7.3.2.deb
$ sudo apt-mark hold logstash
#7.4.0以降でエラー修正が行われたら、下記コマンドでバージョン固定を解除します。
#sudo apt-mark unhold logstash

確認はこちら

$ dpkg --get-selections | grep hold
logstash                                        hold