OSSでNetFlowを解析しよう！ElastiFlow

## はじめに
[ElastiFlow](https://github.com/robcowart/elastiflow)というOSSをあなたはご存知でしょうか。
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/577f96d5-3672-dc4c-78ad-d1e650a7d4e6.png)
NetFlowやsFlow等のフロー情報は通常有料の製品で解析されているかと思います。
製品では[NetFlowAnalyzer](https://www.manageengine.jp/products/NetFlow_Analyzer/)や[FlowMon](https://www.orizon.co.jp/products/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インデックス作成におけるパフォーマンス考慮事項](https://www.elastic.co/jp/blog/performance-considerations-elasticsearch-indexing)より引用
ES JVM Heap + LS JVM Heap = 搭載メモリの半分が目安です。

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

```bash
$ 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を通します。

```bash
$ 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が通っている事を確認しましょう。

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

## Elasticsearchのインストール

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

```bash
$ 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
```
下記の通り追記します。

```bash
$ 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のメモリを必要とするため、確保するように設定します。

```bash
$ sudo vi /etc/elasticsearch/jvm.options
-Xms12g
-Xmx12g
```
設定が終わったら起動します。

```bash
$ sudo systemctl start elasticsearch
$ systemctl status elasticsearch
$ sudo systemctl enable elasticsearch
$ curl "http://localhost:9200/"
```
curlの結果はjson形式で返ってきたら成功です。ちょっと時間かかります。(1分くらい

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

```bash
$ sudo apt install kibana=7.8.1
$ sudo apt-mark hold kibana
```
下記の通り追記します。

```bash
$ sudo vi /etc/kibana/kibana.yml
server.port: 5601
server.host: 0.0.0.0
elasticsearch.hosts: ["http://localhost:9200"]
i18n.locale: "ja-JP"
```
設定が終わったら起動します。

```bash
$ sudo systemctl start kibana
$ systemctl status kibana
$ sudo systemctl enable kibana
```

## Logstashのインストール

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

```bash
$ sudo vi /etc/logstash/jvm.options
-Xms4g
-Xmx4g
```
パケット落ちを防ぐために、Logstash の優先度を上げます。
※これを忘れると非常に処理が遅くなります。

```bash
$ sudo vi /etc/systemd/system/logstash.service
Nice=0
```
設定が終わったら起動します。

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

## ElastiFlowのインストール
※[本家](https://github.com/robcowart/elastiflow)のOSS開発が凍結されているので、私のリポジトリからcloneする手順になっています。

```bash
$ 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`をつけて無効化します。

```bash
$ 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
```
環境に合わせて設定変更していきます。

```bash
$ sudo vi 30_output_10_single.logstash.conf
hosts => [ "${ELASTIFLOW_ES_HOST:localhost:9200}" ]
```
9995ポートで待ち受ける場合。

```bash
$ sudo vi 10_input_netflow_ipv4.logstash.conf
host => "${ELASTIFLOW_NETFLOW_IPV4_HOST:0.0.0.0}"
port => "${ELASTIFLOW_NETFLOW_IPV4_PORT:9995}"
```
systemd内にコピーします。

```bash
$ 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`から選択します。

```bash
$ 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"
```
フローを吐き出す機器が`cisco`or`fortinet`or`velocloud`のいずれかの場合、IPアドレスを設定しておくとElastiFlow上でアプリケーション名が表示されるようになります。

```bash
$ 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"
```
サンプリングを行っている場合はその割合を設定します。

```bash
$ sudo vi /etc/logstash/elastiflow/user_settings/sampling_interval.yml
"192.0.2.1": 1024
"192.0.2.2": 512
```
インタフェースのMIB IDがわかる場合は設定しておくと、その名前で記録されます。

```bash
$ sudo vi /etc/logstash/elastiflow/user_settings/ifName.yml
"192.0.2.1::ifName.0": "eth0"
```
logstashにelastiflowの設定を読み込ませます。

```bash
$ sudo vi /etc/logstash/pipelines.yml
- pipeline.id: elastiflow
  path.config: "/etc/logstash/elastiflow/conf.d/*.conf"
```
設定が終わったら起動します。

```bash
$ 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`も同梱しています。
![import.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/83152b0a-a797-df7c-4f60-7616e7df7183.png)

以上、フローデータが貯れば解析可能となります。お疲れさまでした。
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/e111fa69-b10b-9eed-5425-2e3df8226746.png)
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/54aee9ac-96cc-9a5b-00d6-2e237a7b3ba6.png)

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

```bash
$ 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](https://dev.maxmind.com/geoip/geoip2/geolite2/)を利用しています。
こちらもアップデート用scriptを用意しておきましょう。
DBは毎週火曜日(米国時間)に更新されています。
※有志がGithub上で最新データを公開してくれてるのでそれを使います。

```bash
$ 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`に設定、`ロールオーバーを有効にする`チェックを外し、`削除フェーズを有効にする`チェックを入れます。
削除フェーズのタイミングをインデックスの作成からの経過日数として任意の日数を設定しましょう。
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/696d795b-0d82-f27e-6d78-0cc3cb96862b.png)
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/221440/b66f6fc1-c4db-cf70-6963-b643a998f504.png)

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

```bash
$ 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
```
これで設定完了です。
設定が有効になっているか確認しましょう。

```bash
$ curl http://localhost:9200/_nodes/process?pretty
```
`"mlockall" : true`になっていればOKです。

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

```bash
$ 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にてエラーが修正されているのを確認しました。

```bash
$ 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
```
確認はこちら

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