Kinesis Firehoseを使ってSplunkでAWSユーザーアクティビティを可視化してみた

AWS re:invent 2017でKinesis Firehose経由でのSplunkへのデータ投入ができるようになったと発表されました。

[Amazon Kinesis Data Firehose、配信先として Splunk が一般利用可能に](https://aws.amazon.com/jp/about-aws/whats-new/2017/12/amazon-kinesis-data-firehose-announces-the-general-availability-of-splunk-as-a-destination/)

CloudTrailやVPC Flow、CloudWatch Event、IoTをFirehose経由でストリームできるらしい。

ということで、試してみた。

# はじめに
Splunk: バージョン7.0.2
OS: Amazon Linux on EC2

AWSユーザーアクティビティをモニタリングするには **CloudTrail** から証跡情報をとる必要があります。

データの流れは↓
CloudTrail --> CloudWatch Logs --> Kinesis Firehose --> Splunk

（参考）[Power data ingestion into Splunk using Amazon Kinesis Data Firehose](https://aws.amazon.com/jp/blogs/big-data/power-data-ingestion-into-splunk-using-amazon-kinesis-data-firehose/)

# Splunk設定
最初にデータの受け口となるSplunkをセットアップ

## Appインストール
[Splunk Add-on for Amazon Kinesis Firehose](https://splunkbase.splunk.com/app/3719/) をSplunkにインストール

## HTTP Event Collector設定
Firehose用のHECトークンを作成
[設定] > [データ入力] から [HTTPイベントコレクタ] を選択

[新規トークン] をクリック
<img width="253" alt="Screen Shot 2018-02-23 8.19.25 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/b4f758e6-8733-8ffa-f366-9ab2f70bdcfa.png">

名前を入力して **インデクサー確認を有効化** にチェック
HEC入力設定ごとにソースタイプを決めることになるので、CloudTrailからのデータとわかるよう `aws_kinesis_firehose_cloudtrail` と命名
<img width="626" alt="Screen Shot 2018-02-23 8.27.21 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/29a35bf1-3c6a-4ba2-15da-6c9d2aab6f0d.png">


**ソースタイプ** を `aws:cloudtrail` に、 **インデックス** を `main` に設定
（データへのアクセス権を制御したい場合は個別にインデックスを作成）
<img width="578" alt="Screen Shot 2018-02-23 8.23.58 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/2b3b75ee-d513-37e4-ad60-acbc11da08c9.png">

そのまま次の画面で内容確認して問題なければ [実行]

作成したHEC入力のトークン値を控えておく
![Screen Shot 2018-02-23 8.29.52 PM_mosaic.png](https://qiita-image-store.s3.amazonaws.com/0/174837/afe9b3f9-801d-fbcd-70cc-6c28928a4618.png)

## SplunkのHTTPS設定
Amazon Kinesis Firehoseはデータ送信先のエンドポイントが商用の**SSL証明書**で暗号化されていることが必須要件となるらしい。
オレオレはダメみたい。

[前回、Let's Encryptの証明書をSplunkにインストールしてみた](https://qiita.com/kikeyama/items/be380efbc0fc647ef5df) ところ、FirehoseはLet's Encryptを認証してくれませんでした（2018年1月時点）。

今回のSplunkインスタンスはEC2で稼働しているので、前段にELBを入れて、ACM (Amazon Certificate Manager) からAmazon署名の証明書を取得して対処しました。

ターゲットのEC2には、プロトコル **HTTPS** でポート **8088** に転送するようにしました。
<img width="638" alt="Screen Shot 2018-02-23 8.41.25 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/53740665-423c-7a26-0b3a-3911e825ccf8.png">

ELBが受ける通信は **HTTPS** ポート **443** で設定
<img width="667" alt="Screen Shot 2018-02-23 8.42.22 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/55256568-73c3-d2af-5933-8a4ac4190e18.png">

これでSplunk側の設定は完了！

# AWS設定
続いてAWS側の設定
Lambda, Firehose, CloudWatch Logs, CloudTrailの順に設定します。

## Lambda設定
まずはLambdaから

なぜLambdaが必要かって？
なぜなら、CloudWatch Logsから送られるデータは圧縮されているので、それをFirehose経由で送ってもSplunkが受け取れないから、解凍する処理を経由させる必要があるんです。

ちなみに、解凍の処理を入れずにFirehoseからSplunkに流そうとすると、以下のような `InvalidEncodingException` エラーが出ます。

```json

{"attemptsMade":0,"arrivalTimestamp":1517204317073,"errorCode":"InvalidEncodingException","errorMessage":"The data could not be decoded as UTF-8","attemptEndingTimestamp":25314598570688,"rawData":"...","EventId":null}
```

ということで、Lambdaで新しい関数を作成

設計図（Blueprint）から `kinesis-firehose-cloudwatch-logs-processor` を選択して新規作成
検索フィルタに `firehose` `splunk` と入力すると出てきます。

ソースコード

```javascript

/*
For processing data sent to Firehose by Cloudwatch Logs subscription filters.

Cloudwatch Logs sends to Firehose records that look like this:

{
  "messageType": "DATA_MESSAGE",
  "owner": "123456789012",
  "logGroup": "log_group_name",
  "logStream": "log_stream_name",
  "subscriptionFilters": [
    "subscription_filter_name"
  ],
  "logEvents": [
    {
      "id": "01234567890123456789012345678901234567890123456789012345",
      "timestamp": 1510109208016,
      "message": "log message 1"
    },
    {
      "id": "01234567890123456789012345678901234567890123456789012345",
      "timestamp": 1510109208017,
      "message": "log message 2"
    }
    ...
  ]
}

The data is additionally compressed with GZIP.

The code below will:

1) Gunzip the data
2) Parse the json
3) Set the result to ProcessingFailed for any record whose messageType is not DATA_MESSAGE, thus redirecting them to the
   processing error output. Such records do not contain any log events. You can modify the code to set the result to
   Dropped instead to get rid of these records completely.
4) For records whose messageType is DATA_MESSAGE, extract the individual log events from the logEvents field, and pass
   each one to the transformLogEvent method. You can modify the transformLogEvent method to perform custom
   transformations on the log events.
5) Concatenate the result from (4) together and set the result as the data of the record returned to Firehose. Note that
   this step will not add any delimiters. Delimiters should be appended by the logic within the transformLogEvent
   method.
*/

'use strict';

const zlib = require('zlib');

/**
 * logEvent has this format:
 *
 * {
 *   "id": "01234567890123456789012345678901234567890123456789012345",
 *   "timestamp": 1510109208016,
 *   "message": "log message 1"
 * }
 *
 * The default implementation below just extracts the message and appends a newline to it.
 *
 * The result must be returned in a Promise.
 */
function transformLogEvent(logEvent) {
    return Promise.resolve(`${logEvent.message}\n`);
}

exports.handler = (event, context, callback) => {
    Promise.all(event.records.map(r => {
        const buffer = new Buffer(r.data, 'base64');
        const decompressed = zlib.gunzipSync(buffer);
        const data = JSON.parse(decompressed);
        if (data.messageType !== 'DATA_MESSAGE') {
            return Promise.resolve({
                recordId: r.recordId,
                result: 'ProcessingFailed',
            });
        } else {
            const promises = data.logEvents.map(transformLogEvent);
            return Promise.all(promises).then(transformed => {
                const payload = transformed.reduce((a, v) => a + v, '');
                const encoded = new Buffer(payload).toString('base64');
                return {
                    recordId: r.recordId,
                    result: 'Ok',
                    data: encoded,
                };
            });
        }
    })).then(recs => callback(null, { records: recs }));
};
```

## Kinesis Firehose設定
続いてFirehose
Kinesisの設定画面の左側メニューから **Data Firehose** を選択
<img width="198" alt="Screen Shot 2018-02-23 10.16.30 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/ee5c0a31-55b8-96d6-1432-4b1ff3230a6f.png">

**Create Delivery Stream** をクリックして新しいストリームを作成
ストリーム名を入力して
![firehose__01_01.png](https://qiita-image-store.s3.amazonaws.com/0/174837/294dd29f-a881-d67c-d4fa-e2c7394f659e.png)

Sourceは **Direct PUT or other sources** を選択
![firehose_01_02.png](https://qiita-image-store.s3.amazonaws.com/0/174837/941bbfdb-2e44-e51c-d0b0-1fcc118df9aa.png)

**Transform records with AWS Lambda** で作成した関数を指定
![firehose_02_01.png](https://qiita-image-store.s3.amazonaws.com/0/174837/c5f5b6d2-3e5f-7ddb-5be8-4117e2eb9762.png)

IAMロールのポリシーはデフォルトでおk

続いてデータ送信先を指定
![firehose_03_01.png](https://qiita-image-store.s3.amazonaws.com/0/174837/827845fe-e5d5-41fc-a79c-1a0476215b68.png)

S3、Redshift、Elasticsearch とAWSのサービスと同列にサードパーティーの **Splunk**が並んでるのすごい！

で、　Splunkのエンドポイントとトークンを指定
![firehose_03_02_mosaic.png](https://qiita-image-store.s3.amazonaws.com/0/174837/a26dfb4c-6b4d-db5b-9ca1-cbee5665daa1.png)

エンドポイントはELBで設定したFQDNを（ELB介さない場合はポート `:8088` の指定が必要）、トークンは先のステップでSplunkで作成したトークンを指定
**Splunk endpoint type** は `Raw endpoint` を指定

送信失敗した場合の転送先S3バケットは適当に

より早く送信してほしいので、バッファサイズと時間はなるべく小さくしてみた
![firehose_04_01.png](https://qiita-image-store.s3.amazonaws.com/0/174837/059dec7a-6129-7e52-de18-516db3ad31c0.png)

内容確認して作成！

## CloudWatch Logs設定
（参考） [例 3: Amazon Kinesis Data Firehose のサブスクリプションフィルタ](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs//SubscriptionFilters.html#FirehoseExample)

まずはCWLロググループをひとつ `kikeyama/CloudTrail` 作成

その後、SubscriptionでFirehoseに流す

```bash
aws logs put-subscription-filter \
    --log-group-name "kikeyama/CloudTrail" \
    --filter-name "Destination" \
    --filter-pattern "" \
    --destination-arn "arn:aws:firehose:us-west-2:123456789012:deliverystream/kikeyama-kinesis-firehose-cloudtrail" \
    --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
```

`--filter-pattern` は必須のパラメータなので、何かしらのフィルタリングが必要。 `""` 指定で全条件になるっぽい。
フィルタの指定方法 → [CloudWatch Logs サブスクリプションフィルタの使用](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/CreateDestination.html)

フィルタでの絞り込みなしで全データ取り込みたい場合は、 `--filter-pattern ""` でいけるはず。


ちなみにIAMロールのポリシーは↓

```json:CWLtoKinesisFirehoseRole
{
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "firehose:*"
            ],
            "Resource": [
                "arn:aws:firehose:us-west-2:123456789012:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "iam:PassRole"
            ],
            "Resource": [
                "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
            ]
        }
    ]
}
```


-- 2018年2月27日追記 ｺｺｶﾗ --
ロールの信頼関係を編集して権限付与

<img width="605" alt="Screen Shot 2018-02-26 7.32.43 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/923b9fae-3d22-ef88-00eb-24f95f38b69b.png">

```json
{
  "Statement": {
    "Effect": "Allow",
    "Principal": { "Service": "logs.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }
}
```

-- 2018年2月27日追記 ｺｺﾏﾃﾞ --

以上！

## CloudTrail設定
最後に、CloudTrailを作ってオシマイ

CWLの設定に先ほど作ったロググループを指定
<img width="845" alt="Screen Shot 2018-02-23 10.54.54 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/60ec422a-f9ed-7633-f05d-6f338e13f102.png">

IAMは↓

```json:CloudTrail_CloudWatchLogs_Role
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AWSCloudTrailCreateLogStream20141101",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream"
      ],
      "Resource": [
        "arn:aws:logs:us-west-2:123456789012:log-group:kikeyama/CloudTrail:log-stream:123456789012_CloudTrail_us-west-2*"
      ]
    },
    {
      "Sid": "AWSCloudTrailPutLogEvents20141101",
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents"
      ],
      "Resource": [
        "arn:aws:logs:us-west-2:123456789012:log-group:kikeyama/CloudTrail:log-stream:123456789012_CloudTrail_us-west-2*"
      ]
    }
  ]
}
```

# Splunkでサーチ
![Screen Shot 2018-02-23 11.02.12 PM_mosaic.png](https://qiita-image-store.s3.amazonaws.com/0/174837/6fd282bc-fe8d-eb04-0e8e-6d461a9dd619.png)

いいねー、データ入ってきてるねー

# ユーザーアクティビティを可視化
可視化には [Splunk App for AWS](https://splunkbase.splunk.com/app/1274/) を使います。
SplunkにこのAppをインストールして、Appのメニューから **Security** > **User Activity** に進むと、、、
<img width="683" alt="Screen Shot 2018-02-23 11.36.12 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/c730b215-34e9-d7ad-e5ae-0d824c788404.png">

↓↓↓

おっ！
<img width="1246" alt="Screen Shot 2018-02-23 11.38.25 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/174837/843fd309-528c-1c4e-f655-ebb9254f5734.png">

エラーや非認証のアクティビティが集計して可視化されてる！
非認証多いな、なんでだろう...

CloudTrailで **証跡情報を全てのリージョンに適用** にチェック入れてると、全リージョン情報が出てくるし、ダッシュボードでは自分が見たいリージョンだけをフィルタリングして表示することもできるみたい。

# おわりに
この設定、AWS素人の自分にはなかなか難しかったです。

つまずいたポイントは、、、

- FirehoseからSplunkに送信する際に
 `"errorCode":"InvalidEncodingException","errorMessage":"The data could not be decoded as UTF-8"` がS3に吐き出されていたものの、対処法がわからず右往左往
- 各サービスでIAMのロール設定が必要になるけど、どんな権限付与すればいいのかわからない（というより、ポリシーの設定方法がよくわかってなかった）
- FirehoseがLet's Encryptを認証してくれない

ってところですかね。

VPC Flow Logも同じようにCWLからFirehose経由でSplunkに送ることができます。
その際、Splunkに取り込む際の**ソースタイプがCloudTrailとは異なる**ので、**HECトークンもCloudTrailとは別々に作る**必要があります。

うまくいってなかなかの達成感でした。