CloudTrailをElasticsearchに取り込んでみた

## はじめに
AWS CloudTrailのログ分析をElasticsearchで実施するためのログ取り込み方法をまとめてみました。

## 利用環境
| product | version |
|---------|---------|
| Elasticsearch | 6.6.2 |
| kibana | 6.6.2 |
| logstash | 6.6.2 |
| Java | 1.8.0 |
| OS(EC2) | Amazon Linux2 (t3.large) |
| AMI ID | ami-06b382aba6c5a4f2c |
| Region | us-east-1 |

**【構成図】**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/87133a01-0e58-6ce4-e9f2-57b5278c19e5.png)


**【参考】**
・[AWS CloudTrailとは][00]

## 前提として
CloudTrailはS3 BucketとCloudWatchLogsの2か所に出力が可能です。
コスト(蓄積とリクエスト)的には当然S3の方がメリットがあるんですが
複雑な配列構造のJSONをパースすることを考えるとCloudWatchLogsにメリットがあるため
CloudWatchLogsに出力したものをLogstashでInputしています。
また本記事ではElasticStackのインストール手順は省力しています。

**【参考】**
・[Amazon S3がCloudTrailに対応しました][01]
・[Amazon S3の料金][02]
・[Amazon CloudWatchの料金][03]

## 構築手順
以下の手順で構築を実施します。
　1. IAM Role作成
　2. S3 Bucket作成
　3. ロググループ作成
　4. CloudTrail証跡作成
　5. Logstashの設定
　6. Index_Templeteの追加
　7. ログ取り込み

#### 1. IAM Role作成
- LogstashがCloudWatchLogsからログを取得するのに必要なIAM Roleを作成します。

```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. S3 Bucket作成
- S3 Bucketを作成します。(作成手順は割愛します)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/0150893f-d874-7e30-056a-a940bcb45437.png)
※証跡を作成するためにS3 Bucketが必要なため

#### 3. ロググループ作成
- CloudWatchでロググループを作成します。(CloudTrail/AWSLogGroupと名前を付けます)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/e796e9f6-cf0e-8342-a935-bf29bd6045af.png)

#### 4. CloudTrail証跡作成
- CloudTrailの証跡を作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/2d75f3c2-1ac6-3f12-d137-15d766b5d685.png)

- 作成したS3 Bucketを指定し、[作成]ボタンをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/4917e000-0f88-a82c-1714-f8441d6a0ba0.png)

- 作成した証跡を編集し、[CloudWatch Logs]を設定します。(作成したロググループを指定し、[次へ]をクリックします)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/fa3039f1-89bd-6b61-d2ab-c9acd8e363da.png)

- IAMロールに[CloudTrail_CloudWatchLogs_Role]が指定されていることを確認し、[許可]をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/211371/c9bcec17-b1d6-c9f4-dd20-4cf8d41fb818.png)

#### 5. Logstashの設定
- `logstash-input-cloudwatch_logs`pluginを導入します。

```console
[root@ip-172-31-34-49]# /usr/share/logstash/bin/logstash-plugin install logstash-input-cloudwatch_logs
Validating logstash-input-cloudwatch_logs
Installing logstash-input-cloudwatch_logs
Installation successful
```

- 以下の設定をlogstash.confにします。

```json:/etc/logstash/conf.d/logstash.conf
input {
  cloudwatch_logs {
    region => "us-east-1"
    log_group => [ "CloudTrail/AWSLogGroup" ]
    sincedb_path => "/var/lib/logstash/sincedb_cloudtrail"
  }
}

filter {
  json {
    source => "message"
  }
  date {
    match => [ "eventTime", "ISO8601" ]
    target => "@timestamp"
  }
  ruby {
    code => "event.set('[@metadata][local_time]',event.get('[@timestamp]').time.localtime.strftime('%Y-%m-%d'))"
  }
  useragent {
    source => "userAgent"
    target => "useragent"
  }
  geoip {
    source => "sourceIPAddress"
  }
  mutate {
    remove_field => [ "message" ]
  }
}

output {
  elasticsearch {
    hosts => [ "localhost:9200" ]
    index => "cloudtrail-%{[@metadata][local_time]}"
  }
}
```


#### 6. Index_Templeteの追加
- CloudTrailのログを取り込むとField数がデフォルト制限値の1,000を超えてIndex出来なくなる場合があります。
- 以下のIndex TempleteをKibanaのDevToolsなどでElasticsearchにPUTしておきます。

```json:template/cloudtrail
PUT _template/cloudtrail
{
"index_patterns": ["cloudtrail-*"],
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas" : 1,
    "index.mapping.total_fields.limit": 2000
  },
  "mappings" : {
    "_default_" : {
      "dynamic_templates" : [
        {
          "message_field" : {
            "path_match" : "message",
            "match_mapping_type" : "string",
            "mapping" : {
              "type" : "text",
              "norms" : false
            }
          }
        },
        {
          "string_fields" : {
            "match" : "*",
            "match_mapping_type" : "string",
            "mapping" : {
              "type" : "text",
              "norms" : false,
              "fields" : {
                "keyword" : {
                  "type" : "keyword",
                  "ignore_above" : 256
                }
               }
            }
          }
        }
       ],
       "properties" : {
          "@timestamp" : {
            "type" : "date"
          },
          "@version" : {
            "type" : "keyword"
          },
          "geoip" : {
            "dynamic" : true,
            "properties" : {
              "ip" : {
                "type" : "ip"
              },
              "location" : {
                "type" : "geo_point"
              },
              "latitude" : {
                "type" : "half_float"
              },
              "longitude" : {
                "type" : "half_float"
              }
            }
          }
        }
      }
    },
  "aliases" : { }
}
```

**【参考】**
・ [ElasticSearchでdocument追加時にfield上限値を超過したエラーが出る][04]

#### 7. ログ取り込み
- LogstashのEC2に作成したIAM Role(CloudWatchLogsReadOnlyAccess)を割り当てます。
- Logstashのサービスを起動します。

```console
[root@ip-172-31-34-49]# systemctl start logstash
```
- 以下のような形でKibanaでログが参照出来るようになります。

```json
{
  "_index": "cloudtrail-2019-04-25",
  "_type": "doc",
  "_id": "QShCUmoBhVwl9UKOhoFB",
  "_version": 1,
  "_score": null,
  "_source": {
    "sourceIPAddress": "xxx.xxx.xxx.xxx",
    "awsRegion": "us-east-1",
    "eventName": "ListServiceSpecificCredentials",
    "requestParameters": {
      "userName": "xxxxx",
      "serviceName": "codecommit.amazonaws.com"
    },
    "eventType": "AwsApiCall",
    "userAgent": "console.amazonaws.com",
    "eventTime": "2019-04-25T02:00:38Z",
    "geoip": {
      "ip": "xxx.xxx.xxx.xxx",
      "postal_code": "xxx-xxxxx",
      "timezone": "Asia/Tokyo",
      "region_name": "Tokyo",
      "region_code": "13",
      "country_name": "Japan",
      "latitude": xx.xxx,
      "country_code2": "JP",
      "longitude": xxx.xxxx,
      "city_name": "Tokyo",
      "location": {
        "lat": xx.xxx,
        "lon": xxx.xxxx
      },
      "continent_code": "AS",
      "country_code3": "JP"
    },
    "requestID": "ec37f520-xxxx-xxxx-xxxx-df78061586e9",
    "@timestamp": "2019-04-25T02:10:08.266Z",
    "eventVersion": "1.05",
    "userIdentity": {
      "arn": "arn:aws:sts::xxxxxxxxxxxx:assumed-role/xxxxxx/xxxxxxxxxxxxxxx",
      "sessionContext": {
        "attributes": {
          "creationDate": "2019-04-25T01:54:13Z",
          "mfaAuthenticated": "xxxxxx"
        },
        "sessionIssuer": {
          "principalId": "AROAJB3AZKZTZDX3EBCPE",
          "arn": "arn:aws:iam::xxxxxxxxxxxx:role/xxxxxx",
          "accountId": "xxxxxxxxxxxx",
          "userName": "xxxxxx",
          "type": "Role"
        }
      },
      "type": "AssumedRole",
      "accessKeyId": "xxxxxxxxxxxxxxxxxxx",
      "principalId": "xxxxxxxxxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxx",
      "accountId": "xxxxxxxxxxxx"
    },
    "useragent": {
      "name": "Other",
      "device": "Other",
      "os_name": "Other",
      "build": "",
      "os": "Other"
    },
    "cloudwatch_logs": {
      "ingestion_time": "2019-04-25T02:10:08.303Z",
      "log_stream": "xxxxxxxxxxxx_CloudTrail_us-east-1",
      "event_id": "xxxxxxxxxxxx1142016965745979308228964826353176462615707659",
      "log_group": "CloudTrail/AWSLogGroup"
    },
    "recipientAccountId": "xxxxxxxxxxxx",
    "@version": "1",
    "userid": "xxxxxxxxxxxx",
    "eventSource": "iam.amazonaws.com",
    "prinsipalid": "xxxxxxxxxxxxxxxxxxx",
    "responseElements": null,
    "eventID": "xxxxxxxx-xxxx-xxxx-xxxx-6bd2000d98c3",
    "eventCategory": "Other"
  },
  "fields": {
    "eventTime": [
      "2019-04-25T02:00:38.000Z"
    ],
    "cloudwatch_logs.ingestion_time": [
      "2019-04-25T02:10:08.303Z"
    ],
    "@timestamp": [
      "2019-04-25T02:10:08.266Z"
    ],
    "userIdentity.sessionContext.attributes.creationDate": [
      "2019-04-25T01:54:13.000Z"
    ]
  },
  "sort": [
    1556158208266
  ]
}
```

## まとめ
AWSマネージメントコンソール上ではとてもじゃないですが、ログ分析は出来ません。
CloudWatchLogs Insightでクエリーを投げる方法もあると思いますが
CloudTrailのログは`eventName`によって、ログのFieldが異なるため
まずはKibanaでどんなログが入っているのか概要を抑えたいところですよね^^

CloudTrailではAWSマネージメントコンソールへのログイン監査に必要な情報が
記録されています。社内のセキュリティポリシーを守っているのかを抑えるのに
本投稿が役に立つことを願っています。

[00]: https://docs.aws.amazon.com/ja_jp/awscloudtrail/latest/userguide/cloudtrail-user-guide.html "AWS CloudTrailとは"
[01]: https://dev.classmethod.jp/cloud/aws/cloudtrail-integrate-s3/ "Amazon S3がCloudTrailに対応しました"
[02]: https://aws.amazon.com/jp/s3/pricing/ "Amazon S3の料金"
[03]: https://aws.amazon.com/jp/cloudwatch/pricing/ "Amazon CloudWatchの料金"
[04]: https://boomin.yokohama/archives/50 "ElasticSearchでdocument追加時にfield上限値を超過したエラーが出る"