ペルソナ5にドハマリし、先日100時間以上かけてクリアした @tmknom です。 主人公の名前は吉良吉影。怪盗団の名前はキラークイーンです。ちなみに二周目に突入しました。
さて、みなさんもご存知の通り、 AWS ユーザは常にある不安を抱いています。 AWS アカウントを乗っ取られたらどうしよう。バイツァ・ダストよろしく、すべてを吹っ飛ばされたらどうしよう。という不安です。
しかし、我々は AWS を使わずにはいられないという『サガ』を背負ってはいますが『幸福に生きてみせるぞ！』という強い意思を持たねばならないのです。
そんなわけで、意思を現実のものへと体現すべく AWS コンソールへ、いつ誰がサインインしたのかSlackに通知して、 不正アクセスを簡単に検知できる仕組みを構築してみましょう。
- はじめに
- 事前準備
- バージニア北部へリージョンを切り替え
- Lambda の初期セットアップ
- CloudWatch Events の設定
- CloudWatch Logs の設定と確認
- Slack で Incoming Webhooks の URL の払い出し
- Lambda で Slack へ通知するよう修正
- KMS でマスターキーを作成
- AWS CLI で秘密情報を暗号化
- Lambda の最終形を実装
- まとめ
- We're Hiring！
はじめに
ココで構築する仕組みは「Lambda + CloudWatch Events + KMS」で実現します。 CloudWatch Events で AWS コンソールへのサインインを検知して、それを Lambda 経由で Slack に通知します。 KMS には、Slack 通知に必要な秘密情報を Lambda に持たせる際に使用します。
事前準備
最低限のセキュリティ設定として IAM User と CloudTrail の設定はしておきましょう。以前筆者が Qiita に書いたAWSアカウントを取得したら速攻でやっておくべき初期設定まとめあたりを参考にしてもらえれば、必要なことはすべて載っています。
また、後半で AWS CLI が出てくるので、インストールがまだの人は、公式ドキュメントを参考にインストールしておいてください。
準備 OK の人は早速やってみましょう。
バージニア北部へリージョンを切り替え
いきなり注意点ですが、AWS コンソールへのサインインイベントはバージニア北部リージョン（us-east-1）にしか通知されません。
なので、普段、東京リージョンなどを使っている人は最初にリージョンを切り替えましょう。
Lambda の初期セットアップ
CloudWatch Events から Lambda をキックするために、事前に Lambda Function を作成します。
Lambda のページを開く
「Get Started Now」ボタンをクリック（バージニア北部リージョンではじめて Lambda を作る場合）
Runtime で「Python2.7」を選択し、blueprint の「hello-world-python」をクリック
そのまま「Next」ボタンをクリック
Lambda function の設定をする
- 任意の名前と概要を入力
- Name : sign-in-event-monitor-python
- Description : Sign-in event monitoring
- Lambda function handler and role で新しい Role の作成
- Role で「Create a custom role」を選択
- Role で「Create a custom role」を選択
- IAM 設定ページが別タブで自動で開くので、そのまま「許可」ボタンをクリック
- これで lambda_basic_execution という Role が自動作成される
- これで lambda_basic_execution という Role が自動作成される
- 先程の Role を選択し、ついでにタイムアウト時間を変更して、「Next」ボタンをクリック
- Role : Choose an existing role
- Existing role : lambda_basic_execution
- Timeout：1 min 0 sec
- 任意の名前と概要を入力
入力内容に問題がないことを確認して、「Create function」ボタンをクリック
正常に Lambda Function が作成されたことを確認
作成した Lambda Function の修正
- lambda_handler メソッド内の一行目のコメントアウトを外し、以降の記述をすべて削除
- 9行目の
indent=2を削除（削除しなくても動作に影響はないが、削除したほうがログが見やすい） - コードを修正したら「Save」ボタンをクリック
手順通りやると Lambda Function はこうなります。
from __future__ import print_function
import json
print('Loading function')
def lambda_handler(event, context):
print("Received event: " + json.dumps(event))
CloudWatch Events の設定
CloudWatch Events で AWS コンソールへのサインインを検知するよう設定します。
CloudWatch Events のページを開く
「ルールの作成」ボタンをクリック
ルールを下記のように作成して、「設定の詳細」ボタンをクリック
ルールの名前と説明を入力して、「ルールの作成」ボタンをクリック
- 名前 : sign-in-event-monitor-rule
- 説明 : Sign-in event monitoring rule
先ほど作成したルール（sign-in-event-monitor-rule）があることを確認
CloudWatch Logs の設定と確認
CloudWatch Events から Lambda がキックされていることを CloudWatch Logs で確認してみます。 確認ついでに、ログの保存期間の設定もやっちゃいましょう。
一度サインアウトしてから、もう一度サインインする
- バージニア北部リージョンであることを忘れずに確認
CloudWatch Logs のページを開く
ロググループ「/aws/lambda/sign-in-event-monitor-python」の「失効しない」のリンクをクリック
保存期間を任意の期間（ここでは「１週間」）に変更し、「OK」ボタンをクリック
ログストリームの「20XX/XX/XX/[$LATEST]xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx」のリンクをクリック
Received Event:という行を探しだしてクリック
CloudWatch Events から送られてきたパラメータが JSON で記録されていることを確認
画像だと分かりにくいですが、こんな感じのJSONが記録されていればOKです。
{
"id": "6f87d04b-9f74-4f04-a780-7acf4b0a9b38",
"detail-type": "AWS Console Sign In via CloudTrail",
"source": "aws.signin",
"account": "123456789012",
"time": "2016-01-05T18:21:27Z",
"region": "us-east-1",
"resources": [],
"detail": {
"eventVersion": "1.02",
"userIdentity": {
"type": "Root",
"principalId": "123456789012",
"arn": "arn:aws:iam::123456789012:root",
"accountId": "123456789012"
},
"eventTime": "2016-01-05T18:21:27Z",
"eventSource": "signin.amazonaws.com",
"eventName": "ConsoleLogin",
"awsRegion": "us-east-1",
"sourceIPAddress": "0.0.0.0",
"userAgent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/47.0.2526.106 Safari/537.36",
"requestParameters": null,
"responseElements": {
"ConsoleLogin": "Success"
},
"additionalEventData": {
"LoginTo": "https://console.aws.amazon.com/console/home?state=hashArgs%23&isauthcode=true",
"MobileVersion": "No",
"MFAUsed": "No"
},
"eventID": "324731c0-64b3-4421-b552-dfc3c27df4f6",
"eventType": "AwsConsoleSignIn"
}
}
Slack で Incoming Webhooks の URL の払い出し
Slack の Incoming Webhooks のページを開く
Post to Channel で「#general」を選択し、「Add Incoming WebHooks integration」ボタンをクリック
- 通知先の Channel は後で変更できるので「#general」でなくてもいい
- 通知先の Channel は後で変更できるので「#general」でなくてもいい
「Copy URL」のリンクをクリックして、Webhook URL をクリップボードにコピー後、「Save Settings」ボタンをクリック
Lambda で Slack へ通知するよう修正
下準備が整ったので Lambda を修正していきしょう。やることは3つです。
- CloudWatch Events から送られてきた JSON をパース
- Slack へ通知するメッセージに整形
- Slack に Incoming Webhooks で通知を飛ばす
例えば、こんなコードを作ってみます。
なお、 WEB_HOOK_URL には、先程払い出した Webhook URL を設定してください。 また、 CHANNEL には通知したい任意の Channel を設定してください。
# -*- encoding:utf-8 -*-
import json
import commands
import urllib
WEB_HOOK_URL = 'https://hooks.slack.com/services/your/web_hook_url'
CHANNEL = 'your_channel'
def lambda_handler(event, context):
print("Received event: " + json.dumps(event))
message = create_message(event['detail'])
return notify(message, CHANNEL, WEB_HOOK_URL)
def create_message(detail):
user_name = get_user_name(detail['userIdentity'])
event_time = detail['eventTime']
event_name = detail['eventName']
event_result = detail['responseElements'][event_name]
message = "AWSコンソールへのログインを検知しました\n" \
+ "・ユーザ名：" + user_name + "\n" \
+ "・イベント名：" + event_name + "\n" \
+ "・結果：" + event_result + "\n" \
+ "・発生時刻：" + event_time
return message
def get_user_name(user_identity):
if user_identity['type'] == 'Root':
return 'root'
elif user_identity['type'] == 'IAMUser':
return user_identity['userName']
else:
# RootアカウントとIAM User以外のパターンがあるかどうかわからないが念のため
return json.dumps(user_identity)
def notify(message, channel, web_hook_url):
payload = {
"text": message,
"channel": channel,
"username": "AWSアカウントモニターBot",
"icon_emoji": ":ghost:"
}
escaped_payload = urllib.quote_plus(json.dumps(payload).encode('utf-8'))
curl_command = 'curl -s -X POST -d "payload=%s" %s' % (escaped_payload, web_hook_url)
return commands.getoutput(curl_command)
ほとんど説明することはないですが、IAM User と Root アカウントで地味にデータ構造が異なるので、その点は注意しましょう。（get_user_nameメソッドの部分）
これを保存して、サインアウト&サインインを実行すると、Slackに通知が飛んで来るはずです。余談ですが、二段階認証を設定してると、通知が2つ飛んできます。
こんなのが飛んでくれば、成功です！あとは、メッセージ内容などを好きにカスタマイズして、愛でるだけです。
KMS でマスターキーを作成
このままでも動きますが、Incoming Webhooks の URL がそのまま Lambda にベタ書きされるのは気持ち悪いですね。 GitHub などにそのまま放り込むのはためらわれます。そこで、URL を KMS で暗号化して、Lambda には暗号化済みの文字列を持たせます。
それでは、早速マスターキーを作ってみましょう。
KMS のページを開く
はじめて作る場合、「今すぐ始める」ボタンをクリック
フィルターで「米国東部（バージニア北部）」を選択
「キーの作成」ボタンをクリック
エイリアスと説明を入力して、「次のステップ」ボタンをクリック
- エイリアス： lambda_encryption
- 説明： Lambda encryption
キーの管理者として、自分の IAM User にチェックを入れ、「次のステップ」ボタンをクリック
- ここでは、「nekopunch」という IAM User を管理者にする
- ここでは、「nekopunch」という IAM User を管理者にする
キーポリシーのプレビューを確認して、「完了」ボタンをクリック
キー一覧画面で、先程作成した「lambda_encryption」をクリック
キーユーザーの「追加」ボタンをクリック
Lambda 用に作成した Role（lambda_basic_execution）にチェックを入れて、「アタッチ」ボタンをクリック
AWS CLI で秘密情報を暗号化
それでは Incoming Webhooks の URL を AWS CLI で実際に暗号化してみましょう。なお、ここで一つ注意点です。
それは、指定するURLから「https:」を削除することです。
AWS CLI の仕様により、kms encrypt の --plaintext に URL をそのまま書くと、その URL へ HTTP リクエストを飛ばして、リモートにあるファイルを取得しにいこうとします。このドキュメントのRemote Filesの項目にシレッと書いてありますが、最初盛大にハマりました。
$ aws kms encrypt --region us-east-1 --output text --query CiphertextBlob --key-id alias/lambda_encryption --plaintext //hooks.slack.com/services/your/web_hook_url AQECAHismBxxxxxxxxxxxxx
こんな感じでコマンドを実行すると、暗号化された文字列が出てくるので、Lambda ではこれを使いましょう。
Lambda の最終形を実装
sign-in-event-monitor-python に下記のようなメソッドを追加してみましょう。 「https:」なしの文字列を暗号化しているので、復号時に手動で付与するのを忘れずに行います。
def get_web_hook_url(web_hook_url_encrypted):
return 'https:' + decrypt(web_hook_url_encrypted)
def decrypt(encrypted):
import boto3
import base64
return boto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(encrypted))['Plaintext']
Lambda Function 全体としてはこんな感じになります。
# -*- encoding:utf-8 -*-
import json
import commands
import urllib
WEB_HOOK_URL_ENCRYPTED = 'AQECAHismBxxxxxxxxxxxxx' # 暗号化したWEB_HOOK_URL
CHANNEL = 'your_channel'
def lambda_handler(event, context):
print("Received event: " + json.dumps(event))
message = create_message(event['detail'])
# 修正部分
web_hook_url = get_web_hook_url(WEB_HOOK_URL_ENCRYPTED)
return notify(message, CHANNEL, web_hook_url)
def create_message(detail):
user_name = get_user_name(detail['userIdentity'])
event_time = detail['eventTime']
event_name = detail['eventName']
event_result = detail['responseElements'][event_name]
message = "AWSコンソールへのログインを検知しました\n" \
+ "・ユーザ名：" + user_name + "\n" \
+ "・イベント名：" + event_name + "\n" \
+ "・結果：" + event_result + "\n" \
+ "・発生時刻：" + event_time
return message
def get_user_name(user_identity):
if user_identity['type'] == 'Root':
return 'root'
elif user_identity['type'] == 'IAMUser':
return user_identity['userName']
else:
# RootアカウントとIAM User以外のパターンがあるかどうかわからないが念のため
return json.dumps(user_identity)
# 追加部分
def get_web_hook_url(web_hook_url_encrypted):
return 'https:' + decrypt(web_hook_url_encrypted)
# 追加部分
def decrypt(encrypted):
import boto3
import base64
return boto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(encrypted))['Plaintext']
def notify(message, channel, web_hook_url):
payload = {
"text": message,
"channel": channel,
"username": "AWSアカウントモニターBot",
"icon_emoji": ":ghost:"
}
escaped_payload = urllib.quote_plus(json.dumps(payload).encode('utf-8'))
curl_command = 'curl -s -X POST -d "payload=%s" %s' % (escaped_payload, web_hook_url)
return commands.getoutput(curl_command)
ちょっと長いですが、Incoming Webhooks の URL も暗号化されているので、バージョン管理してしまっても大丈夫そうですね。
まとめ
個人的には、 Lambda + CloudWatch Events + KMS というのは強力なコンボだと思っていますが、 案外これらを組み合わせて使う情報が少ないので、取り上げてみました。
KMS は地味な存在ですが、Lambda で秘密情報を扱う場合は、事実上必須のコンポーネントです。 また、CloudWatch Events も定期実行ができるので cron 代わりに使えたりして、実はポテンシャルが高いです。 たとえば、サインイン可能な IAM User が二段階認証を設定しているか定期的にチェックする、 なんてことが本記事で登場したコンポーネントだけで実現可能です。
ちなみに、本記事は、ハンズオンでそのまま使えることを目指して書きました。 すでにバリバリ使いこなしてるよ！って人も、ぜひ社内勉強会などの啓蒙活動に活用してもらえればと思います。
We're Hiring！
クラウドソーシングのクラウドワークスでは、ペルソナ5とジョジョ第4部について語り合えるエンジニアを募集中です。
興味のある方はお寿司ランチを無料で食べながらお話してみませんか？
