Route53のDNSクエリログをAmazonESに取り込んでみた

## はじめに
Amazon Route53で作成したホストゾーンに対する**DNSクエリログ**を
**Amazon Elasticsearch Service**に取り込んみました。


**【参考】**
・ [Amazon Route53とは][00]
・ [DNSクエリのログ記録][01]

## 利用環境
| product | version |
|---------|---------|
| logstash | 7.5.2 |
| Java | 11.0.6 |
| OS(EC2) | Amazon Linux2 (t3.small) |
| AMI ID | ami-011facbea5ec0363b |
| Amazon ES | 7.1 (latest) |
| Region | us-east-1 |
※投稿時点における最新版を採用しています。

**【構成図】**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/1411618e-e5ef-3e14-68fd-0f040a8ba068.png)

## 前提条件
- Elasticsearchへのログ出力にはElastic社のETLツールである**Logstash**を利用しています。
- Amazon Elasticsearch Service(以下、Amazon ES)は、**パブリックアクセス**としています。  
  (IPアドレスによるホワイトリスト制御)

**【参考】**
・ [Logstashとは][02]

## 実施内容
1. IAM Role作成
2. クエリログ記録の設定
3. Amazon ESのドメイン作成
4. Logstashの構築
5. Kibanaでの各設定

### 1. IAM Role作成
- EC2として構築するLogstashに割り当てるIAM Roleを作成します。
- logatashに割り当てるIAM Roleとして`logstashinput`というロールを作成し、  
`CloudWatchLogsReadOnlyAccess`というIAM Policyを割り当てます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/44a5c573-c5fa-c156-9335-6e988a4bc123.png)

```json:CloudWatchLogsReadOnlyAccess
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "logs:Describe*",
                "logs:Get*",
                "logs:List*",
                "logs:StartQuery",
                "logs:StopQuery",
                "logs:TestMetricFilter",
                "logs:FilterLogEvents"
            ],
            "Effect": "Allow",
            "Resource": "*"
        }
    ]
}
```

### 2. クエリログ記録の設定
- [Route53] > [ホストゾーン]をクリックします。
- ログ出力したいドメイン名を選択し、`クエリログ記録の設定`をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/ee8243f4-5fae-1f35-7751-d573e00481c4.png)

- 下記の内容で`CloudWatchログのロググループの設定`を行います。

**【CloudWatchログのロググループの設定】**

|項目 | 値 |
|---------|---------|
|ホストゾーン | <指定したドメイン名> |
|*にクエリログを送信する | 米国東部（バージニア北部）に新しいロググループ |
|新しいロググループ | /aws/route53/<ドメイン名> |
※新しいロググループ名は任意で好きなように設定可能です。
※クエリログの出力先CloudWatchLogsリージョンは、`米国東部（バージニア北部）`のみです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/7109913d-9c23-3521-9fde-b6f039a3aa68.png)

- 下記の内容で`アクセス権限の付与とクエリログ記録の設定`を行います。

**【アクセス権限の付与とクエリログ記録の設定】**

|項目 | 値 |
|---------|---------|
|CloudWatchログのロググループ | /aws/route53/<ドメイン名> |
|CloudWatchログのリソースポリシー | 米国東部（バージニア北部）に新しいリソースポリシーを作成 |
|リソースポリシーの名前 | route53-query-logging-policy |
|リソースポリシーを適用するロググループ | /aws/route53/* |
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/2ac6c5c5-9466-29c1-29ba-c9d668a6caf9.png)

- `ポリシーとテストのアクセス権限の作成`をクリックし、下記の成功メッセージが表示されればOKです。
- `クエリログ記録コンフィグの作成`をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/eff44313-3a88-d245-1ad5-c5006696ecda.png)

- DNSクエリがあると、CloudWatch Logsの作成したロググループ内に 
  Route53のホストゾーンID名のログストリームが作成されていきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/1255caa3-19fe-4aaf-5d09-6bf17fc9060b.png)

- DNSクエリログのフォーマットは以下の通りです。

|項目 | フィールド名 | 説明 | 値 |
|---------|---------|---------|---------|
| ログ形式バージョン | log_version | クエリログのバージョン番号 | 1.0 |
| クエリのタイムスタンプ | timestamp | Route53がリクエストに応答した日時(ISO8601形式) | 2020-02-23T15:50:46Z |
| ホストゾーンID | hostzone_id | Route53のホストゾーンID | Z1CV60OKBEMYL2 |
| クエリ名 | query_name | リクエストで指定されたドメインまたはサブドメイン | xxxxxx.name |
| クエリタイプ | query_type | リクエストで指定されたDNSレコードタイプ | MX |
| レスポンスコード | response_code | Route53が返すDNSレスポンスコード | NOERROR |
| レイヤー 4 プロトコル | protocol | クエリの送信に使用されたプロトコル (TCP/UDP) | UDP |
| Route 53 エッジロケーション | edge_location | クエリに応答したRoute 53エッジロケーション | ORD52-C2 |
| リゾルバー IP アドレス | resolver_ipaddress | DNSリゾルバーのIPアドレス | xxx.215.xx.87 |
| EDNS クライアントサブネット | client_subnet | DNSリゾルバーのサブネット | - |

**【参考】**
・ [Route53でクエリログが取得できるようになりました][03]

### 3. Amazon ESのドメイン作成
- 下記内容でAmazon ESのドメインを作成します。

|項目 | 値 |
|---------|---------|
| リージョン | us-east-1 |
| デプロイタイプ | 開発およびテスト |
| Elasticsearchのバージョン | 7.1 (latest) |
| Elasticsearchのドメイン | loganalytics-dev |
| インスタンスタイプ | c5.large.elasticsearch |
| ノードの数 | 1 |
| データノードのストレージタイプ | EBS |
| EBS ボリュームタイプ | 汎用(SSD) |
| EBS ボリュームサイズ | 10 GiB |
| 自動スナップショットの開始時間 | 00:00 UTC (デフォルト) |
| ネットワークアクセス | パブリックアクセス |
| 細かいアクセスコントロールを有効化 | 無効 |
| Amazon Cognito認証を有効化 | 無効 |
| ドメインアクセスポリシー | カスタムアクセスポリシー (IPv4アドレス) |
| ドメインへのすべてのトラフィックにHTTPSを要求 | 有効 |
| ノード間の暗号化 | 無効 |
| 保管時のデータの暗号化の有効化 | 無効 |
※アクセスポリシーに追加するIPは、Amazon ESにアクセスする外部IPとLogstashのIPになります。

### 4. Logstashの構築
- [Logstash][04]をEC2 Amazon Linux2 (t3.small)で構築します。
- 下記のパイプライン構成ファイル(`/etc/logstash/conf.d/logstash.conf`)の設定を行います。  
  (このタイミングではLogstashは起動しません。)

```json:logstash.conf
input {
  cloudwatch_logs {
    log_group => [ "/aws/route53/<ドメイン名>" ]
    region => "us-east-1"
    interval => 5
  }
}

filter {
  ### dissect filterでフィールドを抽出
  dissect {
      mapping => {
        "message" => "%{log_version} %{timestamp} %{hostzone_id} %{query_name} %{query_type} %{response_code} %{protocol} %{edge_location} %{resolver_ipaddress} %{client_subnet}"
      }
    }
  ### dateフィールドから@timestampを抽出
  date {
    match => [ "timestamp", "ISO8601" ]
    timezone => "Asia/Tokyo"
    target => "@timestamp"
  }
  ### @timstampから日本時間を抽出
  ruby {
    code => "event.set('[@metadata][local_time]',event.get('[@timestamp]').time.localtime.strftime('%Y-%m-%d'))"
  }
  ### グローバルIPから国などをマッピングするために指定
  geoip {
    source => "resolver_ipaddress"
    target => "src_geoip"
    tag_on_failure => "src_geoip_lookup_failure"
  }
  ### document_idに利用する一意のIDを作成
  fingerprint {
    source => "message"
    target => "[@metadata][fingerprint]"
    method => "MURMUR3"
  }
  ### デフォルトの型がstringのため、フィールド定義で定義した型に変換
  mutate {
    ### typeフィールドを追加
    add_field => { "type" => "route53" }    
    ### 不要なフィールドを削除
    remove_field => [ "timestamp" ]
  }
}

output {
  ### 出力先のAmazonESのIndexを指定
  elasticsearch {
    hosts => [ "https://search-loganalytics-dev-xxxxxxxxxxxxxxxxxxxxxx.us-east-1.es.amazonaws.com:443" ]
    index => "%{type}-%{[@metadata][local_time]}"
    document_id => "%{[@metadata][fingerprint]}"
    ilm_enabled => false
  }
}
```

**【2020/4/21追記】**
※ `ilm_enabled => false`は、Amazon ESのバージョン7以降を利用する場合は、記載しないとエラーになるので注意してください。 
※ Logstash 7.0.0で[ILM(Index Lifecycle Management)の自動検出機能を追加][13]され、これが悪さをするようです。

**【参考】**
・ [cloudwatch-logs input][05]
・ [dissect filter][06]
・ [date filter][07]
・ [ruby filter][08]
・ [geoip filter][09]
・ [fingerprint filter][10]
・ [mutate filter][11]
・ [elasticsearch output][12]

### 5. Kibanaでの各設定
- Amazon ESの`Kibana`のURLをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/56a2dd69-c276-cf6b-cae8-18382c74db72.png)

- [Dev Tools]のConsoleからRoute53のIndex Templateを追加します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/5e69b719-12f2-2ffc-edba-f521cf25dcc3.png)

- 上記で張り付けたIndex Templateは以下の通りです。

```json:index_templete_route53
PUT _template/route53
{
  "index_patterns": ["route53-*"],
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas" : 1
  },
  "mappings": {
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "@version" : {
        "type" : "keyword"
      },
      "log_version" : {
        "type" : "keyword"
      },
      "hostzone_id" : {
        "type" : "keyword"
      },
      "query_name" : {
        "type" : "keyword"
      },
      "query_type" : {
        "type" : "keyword"
      },
      "response_code" : {
        "type" : "keyword"
      },
      "protocol" : {
        "type" : "keyword"
      },
      "edge_location" : {
        "type" : "keyword"
      },
      "resolver_ipaddress" : {
        "type" : "ip"
      },
      "client_subnet" : {
        "type" : "keyword"
      },
      "type" : {
        "type" : "keyword"
      },
      "tags" : {
        "type" : "keyword"
      },
      "message" : {
        "type" : "text"
      },
      "cloudwatch_logs" : {
        "properties" : {
          "event_id" : {
            "type" : "keyword"
          },
          "ingestion_time" : {
            "type" : "date"
          },
          "log_group" : {
            "type" : "keyword"
          },
          "log_stream" : {
            "type" : "keyword"
          }
        }
      },
      "src_geoip" : {
        "properties" : {
          "city_name" : {
            "type" : "keyword"
          },
          "continent_code" : {
            "type" : "keyword"
          },
          "country_code2" : {
            "type" : "keyword"
          },
          "country_code3" : {
            "type" : "keyword"
          },
          "country_name" : {
            "type" : "keyword"
          },
          "dma_code" : {
            "type" : "long"
          },
          "ip" : {
            "type" : "ip"
          },
          "latitude" : {
            "type" : "float"
          },
          "location" : {
            "properties" : {
              "lat" : {
                "type" : "float"
              },
              "lon" : {
                "type" : "float"
              }
            }
          },
          "longitude" : {
            "type" : "float"
          },
          "postal_code" : {
            "type" : "keyword"
          },
          "region_code" : {
            "type" : "keyword"
          },
          "region_name" : {
            "type" : "keyword"
          },
          "timezone" : {
            "type" : "keyword"
          }
        }
      }
    }
  }
}
```

- Logstashを起動します。

```console:logstash_start
$ sudo systemctl start logstash
$ sudo systemctl status logstash
● logstash.service - logstash
   Loaded: loaded (/etc/systemd/system/logstash.service; disabled; vendor preset: disabled)
   Active: active (running) since Sun 2020-02-23 17:11:21 UTC; 4min 42s ago
 Main PID: 32168 (java)
   CGroup: /system.slice/logstash.service
           └─32168 /bin/java -Xms1g -Xmx1g -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly -Djava.awt.h...

Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,242][INFO ][logstash.outputs.elasticsearch][main] New Elastic...:443"]}
Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,282][INFO ][logstash.filters.geoip   ][main] Using geoip data....mmdb"}
Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,332][INFO ][logstash.outputs.elasticsearch][main] Using defau...emplate
Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,396][INFO ][logstash.outputs.elasticsearch][main] Attempting ...field"=
Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,460][WARN ][org.logstash.instrument.metrics.gauge.LazyDelegatingGaug...
Feb 23 17:12:00 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:00,467][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pip...
Feb 23 17:12:01 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:01,418][INFO ][logstash.javapipeline    ][main] Pipeline started..."main"}
Feb 23 17:12:01 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:01,497][INFO ][logstash.agent           ] Pipelines running {:co...es=>[]}
Feb 23 17:12:01 ip-172-31-37-204.ec2.internal logstash[32168]: [2020-02-23T17:12:01,736][INFO ][logstash.agent           ] Successfully started L...=>9600}
Feb 23 17:12:14 ip-172-31-37-204.ec2.internal logstash[32168]: /usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-filter-fingerprint-3.2...recated
Hint: Some lines were ellipsized, use -l to show in full.
```
- Kibanaの [Management] > [Index Patterns]で`Create index pattern`をクリックします。
- `route53-*`という名前でIndex Patternを作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/cd14dce3-e217-a227-329d-02a49830f4a2.png)

- `@timestamp`を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/ccf40288-a536-b13e-a1f0-7f043dd0ddf6.png)

- [Discover]を開き、Index Patternに`route53-*`を指定します。
- デフォルトでは直近15分間のログが表示されます。表示されていればOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/e81f6ef3-ac8d-a95d-3213-0b5012307653.png)

- 取り込んだログは以下のような感じです。

```json:route53_log
{
  "_index": "route53-2020-02-23",
  "_type": "_doc",
  "_id": "1366418581",
  "_version": 2,
  "_score": null,
  "_source": {
    "src_geoip": {
      "country_code3": "RU",
      "country_code2": "RU",
      "continent_code": "EU",
      "longitude": 100,
      "ip": "2a00:xxxxxxx:1",
      "location": {
        "lon": 100,
        "lat": 60
      },
      "latitude": 60,
      "country_name": "Russia",
      "timezone": "Asia/Krasnoyarsk"
    },
    "type": "route53",
    "protocol": "UDP",
    "query_type": "MX",
    "response_code": "NOERROR",
    "@version": "1",
    "resolver_ipaddress": "2a00:xxxxxxx:1",
    "log_version": "1.0",
    "edge_location": "CDG3-C2",
    "message": "1.0 2020-02-23T17:08:41Z Z1CV60OKBEMYL2 xxxxxxx.name MX NOERROR UDP CDG3-C2 2a00:xxxxxxx:1 -",
    "hostzone_id": "Z1CV60OKBEMYL2",
    "client_subnet": "-",
    "query_name": "xxxxxxx.name",
    "cloudwatch_logs": {
      "ingestion_time": "2020-02-23T17:09:08.462Z",
      "log_stream": "Z1CV60OKBEMYL2/CDG3-C2",
      "event_id": "35290432438372433059640402101280473034952869042615484416",
      "log_group": "/aws/route53/xxxxxxx.name"
    },
    "@timestamp": "2020-02-23T17:08:41.000Z"
  },
  "fields": {
    "cloudwatch_logs.ingestion_time": [
      "2020-02-23T17:09:08.462Z"
    ],
    "@timestamp": [
      "2020-02-23T17:08:41.000Z"
    ]
  },
  "sort": [
    1582477721000
  ]
}
```

## まとめ
いかがでしたでしょうか。
公開しているWebサービスに対する不審なDNSクエリログを監視することで
攻撃の偵察行為を見つけたり、どこの国からのアクセスがあるのか分析すると
また違った情報が得られるかもしれませんね！

[00]:https://docs.aws.amazon.com/ja_jp/Route53/latest/DeveloperGuide/Welcome.html "Amazon Route53とは"
[01]:https://docs.aws.amazon.com/ja_jp/Route53/latest/DeveloperGuide/query-logs.html "DNSクエリのログ記録"
[02]:https://www.elastic.co/jp/logstash "Logstashとは"
[03]:https://dev.classmethod.jp/cloud/aws/query-log-from-route53/ "Route53でクエリログが取得できるようになりました"
[04]:https://qiita.com/hssh2_bin/items/cbc9408f03fd81b1863a "Amazon Corretto 11で構築するLogstash"
[05]:https://github.com/lukewaite/logstash-input-cloudwatch-logs "cloudwatch-logs input"
[06]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-dissect.html "dissect filter"
[07]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-date.html "date filter"
[08]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-ruby.html "ruby filter"
[09]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-geoip.html "geoip filter"
[10]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-fingerprint.html "fingerprint filter"
[11]:https://www.elastic.co/guide/en/logstash/7.5/plugins-filters-mutate.html "mutate filter"
[12]:https://www.elastic.co/guide/en/logstash/7.5/plugins-outputs-elasticsearch.html "elasticsearch output"
[13]:https://www.elastic.co/jp/blog/logstash-7-0-0-released "Logstash 7.0.0がリリースされました"