fluentdとKibanaでSSHアクセス元マップ


大学外にポートが空いている研究室のサーバに対して，不正なSSHのアクセスが結構来ています．
今回は以下のステップで，不正アクセス元の国をKibana mapで可視化してみました．

  1. fluentdでSSHのログから不正アクセスのIPアドレスを解析
  2. GeoIPを使ってIPアドレスから国を特定
  3. Elasticsearchにログを収集
  4. Kibanaのmap panelを使って可視化

## GeoIPとは

[サイトアクセス者の位置情報を取得するGeoIP | SourceForge.JP Magazine](http://sourceforge.jp/magazine/07/07/17/0047254)より引用．

> GeoIPとは、IPアドレスを国、都市、インターネットサービスプロバイダ(ISP)にマッピングしたデータベース群である。 
> そうしたデータとともにC、PHP、Javaその他いくつかの言語を使ってデータベースにアクセスするためのLGPLライセンスのAPIも用意されている。 

## fluentdでSSHログ解析

fluentd(td-agent)については[前回](http://qiita.com/hiconyan/items/efa8fc888dc91299633e)書いたので割愛します．

SSHのログは，以下のような形式になっていると思います．
ここでは，Invalid user(存在しないユーザ)のログを抽出します．

```:/var/log/auth.log
Feb  8 07:47:28 air sshd[29484]: Invalid user admin from 58.206.126.29
```

fluentdの設定は以下の通り．
formatでは，時刻とユーザ名，IPアドレスを取っています．

```apacheconf:/etc/td-agent/td-agent.conf
<source>
  type tail
  format /^(?<time>[^0-9]+ [^ ]+ [^ ]+) .+ sshd\[.+\]: Invalid user (?<user>[^ ]+) from (?<ip_address>[^$]+)$/
  time_format %b %d %H:%M:%S
  path /var/log/auth.log
  pos_file /tmp/auth.log.pos
  tag geo.ssh.access
</source>
```

## fluent-geoip-pluginの導入

fluentd用にgeoipプラグインを作られている方が居ました．

  - [fluent-plugin-geoip v0.0.4 をリリースしました。ElasticSearch＋Kibanaの世界地図に位置情報をプロットするために必要なFluentdの設定サンプルも紹介します - Y-Ken Studio](http://y-ken.hatenablog.com/entry/fluent-plugin-geoip-v0.0.4)
  - [プレースホルダ設定に対応した fluent-plugin-geoip v0.1.1 をリリースしました - Y-Ken Studio](http://y-ken.hatenablog.com/entry/fluent-plugin-geoip-v0.1.1)

今回は，ログのIPアドレスから`country_code`(JP，USなど)を取得します．
fluentdの設定に以下の内容を追記します．

```apacheconf:/etc/td-agent/td-agent.conf
<match geo.ssh.*>
  type geoip
  geoip_lookup_key ip_address
  <record>
    country ${country_code['ip_address']}
  </record>
  remove_tag_prefix geo.
  add_tag_prefix host.
</match>
```

## Elasticsearchに投入

fluent-plugin-elasticsearchについては[前回](http://qiita.com/hiconyan/items/efa8fc888dc91299633e)書いたので割愛します．
今回使用した設定は下記の通りです．

```apacheconf:/etc/td-agent/td-agent.conf
<match host.ssh.*>
  type elasticsearch
  host 133.92.147.250
  port 9200
  type_name ssh-invalid
  include_tag_key true
  tag_key @log_name
  logstash_format true
  logstash_prefix ssh-invalid
  flush_interval 10s
</match>
```

インデックスのMappingは自動で問題無いと思いますが，手動で行いました．

```sh:mapping.sh
curl -XPUT http://127.0.0.1:9200/_template/ssh-invalid -d '{
  "template": "ssh-invalid-*",
  "mappings": {
    "_default_": {
      "numeric_detection" : true,
      "properties": {
        "@timestamp": { "type":"date", "format":"dateOptionalTime" },
        "@log_name": {type: "string"},
        "user": {type: "string"},
        "ip_address": {type: "string"},
        "country": {type: "string"}
      }
    }
  }
}'
```

## Kibanaのmapで可視化

Kibanaには，世界地図などにマッピングできるmap panelがあります．

  - [Map](http://www.elasticsearch.org/guide/en/kibana/current/_map.html)

今回は世界地図を使って，`country`の値でマッピングします．
作ったDashboardはGitHubで公開しています．

  - [ansible-playbooks/ssh-invalid.json at master · hico-horiuchi/ansible-playbooks](https://github.com/hico-horiuchi/ansible-playbooks/blob/master/kibana/files/ssh-invalid.json)

実際の画面はこんな感じになりました．
某国だけでなく，アジアや南米からも多くアクセスがあることが分かりますね．

ユーザ名のランキングも作ってみたら，adminがダントツでした．
(rootは存在するユーザなので，ここでは含まれない．)

![kibana_map.png](https://qiita-image-store.s3.amazonaws.com/0/58742/fa1a60a9-e450-9445-6515-81997b22ea24.png "kibana_map.png")