はじめに
目的
AWS Foundational Security Best Practicesに含まれているコントロールである、
[S3.5] S3 汎用バケットでは、SSL を使用するリクエストが必要ですに自動で対応する。
動機
-
cdk bootstrap
で作成されたS3バケットには、
このコントロールに対応したバケットポリシーがアタッチされており、
自動で対応されることに感動をおぼえた。 -
コンソールからのS3バケット作成時には、バケットポリシーを設定できない。
作成後に改めてポリシーを設定する必要があり、手間だと思っていた。
以上の理由から、自動化したいと考えました。
構成
流れは以下の通りです
1. 新しいS3バケットが作成される
2. Amazon EventBridgeがこのバケットの作成イベントを検出
3. EventBridgeが、Lambda関数を起動
4. Lambda関数は以下の処理を実行
- イベントのJSONから新しく作成されたS3バケットの名前を抽出
- バケット名に応じた、目的のバケットポリシーを生成
- 生成したポリシーを、該当するS3バケットにアタッチ
簡単な構成なので不要かもしれませんが、構成図は以下のとおりです。
実装
Eventbridge
前提条件
S3バケットの作成イベントを検知するために、CloudTrailを有効化しておく必要があります。
イベントパターン
今回は、S3バケットの作成のみを検知するようにしたいので、下記のとおりです。
イベントパターン
{
"detail-type": ["AWS API Call via CloudTrail"],
"detail": {
"eventSource": ["s3.amazonaws.com"],
"eventName": ["CreateBucket"]
},
"source": ["aws.s3"]
}
Lambda
コード
S3バケットの作成イベントを検知したEventbridgeから渡されたJSONをもとに、
新しく作成されたバケットにSSL通信のみを許可するポリシーを適用します。
主な処理の流れは以下のとおりです。
1.イベント解析
- EventBridgeから送信されたイベントデータから、作成されたS3バケット名を抽出
2.ポリシー生成:
- バケット名に基づいて、SSL通信のみを許可するバケットポリシーを生成
3.ポリシー適用:
- 生成したポリシーを、作成されたS3バケットに適用
4.結果のログ記録:
- ポリシーの適用が成功したか失敗したかを、CloudWatch Logsに記録
Labdaのコード
import boto3
import json
import logging
# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# S3クライアント
s3 = boto3.client('s3')
# バケットポリシーのテンプレート
POLICY_TEMPLATE = {
"Version": "2012-10-17",
"Id": "AccessControl",
"Statement": [
{
"Sid": "AllowSSLRequestsOnly",
"Effect": "Deny",
"Principal": "*",
"Action": "s3:*",
"Resource": [
"arn:aws:s3:::{bucket_name}",
"arn:aws:s3:::{bucket_name}/*"
],
"Condition": {
"Bool": {
"aws:SecureTransport": "false"
}
}
}
]
}
def create_bucket_policy(bucket_name):
# バケット名に基づいてポリシーを生成
policy = json.dumps(POLICY_TEMPLATE).replace("{bucket_name}", bucket_name)
return policy
def lambda_handler(event, context):
# イベントからバケット名を取得
bucket_name = event['detail']['requestParameters']['bucketName']
try:
# バケットポリシーを生成
bucket_policy = create_bucket_policy(bucket_name)
# バケットポリシーを適用
s3.put_bucket_policy(
Bucket=bucket_name,
Policy=bucket_policy
)
logger.info(f"バケットポリシーが正常に適用されました: {bucket_name}")
return {
'statusCode': 200,
'body': json.dumps(f'バケットポリシーが正常に適用されました: {bucket_name}')
}
except Exception as e:
logger.error(f"エラーが発生しました: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'body': json.dumps(f'エラーが発生しました: {str(e)}')
}
pythonは完全に独学で勉強不足なため、良くない部分があるかもしれません。
ご了承ください。
ポリシー
Lambdaの実行ロールは、AWSLambdaBasicExecutionRoleに加え
S3バケットにバケットポリシーをアタッチする許可が必要になります。
追加のIAMポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Action": "s3:PutBucketPolicy",
"Resource": "arn:aws:s3:::*",
"Effect": "Allow"
}
]
}
トリガー
トリガーは、作成したEventbridgeを設定します。
動作確認
バケット名「auto-attach-ssl-policy-test」のS3バケット作成しました。
Lambdaのログは下記のとおりとなっており、想定通り動作したように見えます。
details
timestamp,message
1721046769679,"INIT_START Runtime Version: python:3.12.v28 Runtime Version ARN: arn:aws:lambda:ap-northeast-1::runtime:7776dee4c90fff4d4bf833b78d94e5b4148d14d044b867a14756f301ecfe1e28
"
1721046769964,"[INFO] 2024-07-15T12:32:49.964Z Found credentials in environment variables.
"
1721046770118,"START RequestId: 1c0c1597-7c6e-412e-b51f-54aa0edadf7c Version: $LATEST
"
1721046770985,"[INFO] 2024-07-15T12:32:50.985Z 1c0c1597-7c6e-412e-b51f-54aa0edadf7c バケットポリシーが正常に適用されました: auto-attach-ssl-policy-test
"
1721046770996,"END RequestId: 1c0c1597-7c6e-412e-b51f-54aa0edadf7c
"
1721046770996,"REPORT RequestId: 1c0c1597-7c6e-412e-b51f-54aa0edadf7c Duration: 878.00 ms Billed Duration: 878 ms Memory Size: 128 MB Max Memory Used: 83 MB Init Duration: 437.87 ms
"
バケットポリシーを確認したところ、想定通りのポリシーがアタッチされていました。
アタッチされたバケットポリシー
{
"Version": "2012-10-17",
"Id": "AccessControl",
"Statement": [
{
"Sid": "AllowSSLRequestsOnly",
"Effect": "Deny",
"Principal": "*",
"Action": "s3:*",
"Resource": [
"arn:aws:s3:::auto-attach-ssl-policy-test",
"arn:aws:s3:::auto-attach-ssl-policy-test/*"
],
"Condition": {
"Bool": {
"aws:SecureTransport": "false"
}
}
}
]
}
課題点
本実装には、いくつかの課題や改善の余地があります
1.ポリシー削除への対応
現在の構成では、ポリシーが削除された場合に対応できていません。
この問題に対処するための方法として以下が考えられます。
- EventBridgeでポリシー削除イベントを検知し、再適用するLambda関数の実装
- AWS Config ルールを使用してポリシーの存在を監視し、修復アクションをトリガーする
- AWS Systems Manager オートメーションランブックを使用した定期的なチェックと修復
ただし、上記の方法だと異なるポリシーも適用したうえで削除してしまった場合、
現在のLambdaだと、SSLに関するポリシーのみになってしまいます。
ですので、そもそもポリシーを削除できないように、IAMやSCPで制限すべきだと思います。
2. モニタリングと通知
ポリシー適用の失敗やエラーを適切に監視し、管理者に通知する仕組みが必要です。
(失敗した場合は、SecurityHubで通知されることになるので不要かもしれません。)
方法としては以下が考えられます
- Lambda関数の、errorログを出力している部分でSNS通知も実装する
- CloudWatchLogsのメトリクスフィルタを活用してSNS通知する
- 正常にポリシーがアタッチされるまでリトライするようにしたうえで、
上記2点どちらかのSNS通知を実装する
CDK
CDKのコードを置いておきます。
CDKのコード(python)
from aws_cdk import (
Stack,
aws_lambda as _lambda,
aws_iam as iam,
aws_events as events,
aws_events_targets as targets,
)
from constructs import Construct
class SslLambdaStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Lambda Function
lambda_fn = _lambda.Function(
self, 'SSLLambda',
handler='add_ssl_policy_to_s3.lambda_handler',
code=_lambda.Code.from_asset('lambda'),
runtime=_lambda.Runtime.PYTHON_3_12,
)
# permissions to S3 Bucket
lambda_fn.add_to_role_policy(iam.PolicyStatement(
actions=['s3:PutBucketPolicy'],
resources=['arn:aws:s3:::*'],
))
# Lambda Version
version = lambda_fn.current_version
# EventBridge Rule
rule = events.Rule(
self, 'S3BucketCreationRule',
event_pattern=events.EventPattern(
source=['aws.s3'],
detail_type=['AWS API Call via CloudTrail'],
detail={
'eventSource': ['s3.amazonaws.com'],
'eventName': ['CreateBucket'],
},
)
)
# Add Lambda as target for the EventBridge rule
rule.add_target(targets.LambdaFunction(lambda_fn))
まとめ
この記事では、AWS Foundational Security Best Practicesの
コントロールID[S3.5]に対応するため、S3バケット作成時に自動的に
SSL/HTTPS通信のみを許可するポリシーを適用するシステムを構築しました。
この実装により、コンプライアンス要件の遵守を自動化することができ、
手動によるミスを無くすことが出来ます。
興味のある方は、ぜひ試してみてください。