はじめに

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

【参考】
・ Amazon Route53とは
・ DNSクエリのログ記録

利用環境

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
※投稿時点における最新版を採用しています。

【構成図】

前提条件

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

【参考】
・ Logstashとは

実施内容

IAM Role作成
クエリログ記録の設定
Amazon ESのドメイン作成
Logstashの構築
Kibanaでの各設定

1. IAM Role作成

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

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] > [ホストゾーン]をクリックします。
ログ出力したいドメイン名を選択し、クエリログ記録の設定をクリックします。
下記の内容でCloudWatchログのロググループの設定を行います。

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

項目	値
ホストゾーン	<指定したドメイン名>
*にクエリログを送信する	米国東部（バージニア北部）に新しいロググループ
新しいロググループ	/aws/route53/<ドメイン名>
※新しいロググループ名は任意で好きなように設定可能です。
※クエリログの出力先CloudWatchLogsリージョンは、`米国東部（バージニア北部）`のみです。

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

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

項目	値
CloudWatchログのロググループ	/aws/route53/<ドメイン名>
CloudWatchログのリソースポリシー	米国東部（バージニア北部）に新しいリソースポリシーを作成
リソースポリシーの名前	route53-query-logging-policy
リソースポリシーを適用するロググループ	/aws/route53/*

ポリシーとテストのアクセス権限の作成をクリックし、下記の成功メッセージが表示されればOKです。
クエリログ記録コンフィグの作成をクリックします。
DNSクエリがあると、CloudWatch Logsの作成したロググループ内に
Route53のホストゾーンID名のログストリームが作成されていきます。
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でクエリログが取得できるようになりました

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をEC2 Amazon Linux2 (t3.small)で構築します。
下記のパイプライン構成ファイル(/etc/logstash/conf.d/logstash.conf)の設定を行います。
(このタイミングではLogstashは起動しません。)

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)の自動検出機能を追加され、これが悪さをするようです。

【参考】
・ cloudwatch-logs input
・ dissect filter
・ date filter
・ ruby filter
・ geoip filter
・ fingerprint filter
・ mutate filter
・ elasticsearch output

5. Kibanaでの各設定

Amazon ESのKibanaのURLをクリックします。
[Dev Tools]のConsoleからRoute53のIndex Templateを追加します。
上記で張り付けたIndex Templateは以下の通りです。

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を起動します。

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を作成します。
@timestampを指定します。
[Discover]を開き、Index Patternにroute53-*を指定します。
デフォルトでは直近15分間のログが表示されます。表示されていればOKです。
取り込んだログは以下のような感じです。

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クエリログを監視することで
攻撃の偵察行為を見つけたり、どこの国からのアクセスがあるのか分析すると
また違った情報が得られるかもしれませんね！