AWS LambdaでS3バケットとAPI Gatewayのポリシーを設定してみる

t.honda

2022.06.30

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

AWS Lambdaを使い、S3バケットとAPI GatewayにVPCエンドポイントからのアクセスを許可するポリシーを設定する機会がありました。簡単な処理ではありますが、サンプルとして上げておきたいと思います。

前提条件など

使用する言語はPython、Lambdaのランタイムは「Python 3.7」とします。Lambda関数の作成方法については以下などを参照ください。

Lambda の開始方法

Lambda関数の環境変数に以下の値を登録するものとします。

BUCKET_NAME ・・・ポリシーを設定したいS3バケット名
VPC_ENDPOINT ・・・アクセスを許可したVPCエンドポイント
VPC_ENDPOINT2 ・・・アクセスを許可したVPCエンドポイント（２つ目）
API_GATEWAY_ID ・・・ポリシーを設定したいAPI GatewayのID

Lambdaのソース

Lambdaのソースは以下のようになります。ポリシーについては単純に許可をしているだけなので、Denyなどは適宜加えてください。

import json
import os
import boto3

account_id = boto3.client("sts").get_caller_identity()["Account"]
bucket_name = os.environ["BUCKET_NAME"]
vpc_endpoint = os.environ["VPC_ENDPOINT"]
vpc_endpoint2 = os.environ["VPC_ENDPOINT2"]
api_gateway_id = os.environ["API_GATEWAY_ID"]

def _setup_bucket_policy():
    print("Setup Bucket Policy")
    
    bucket_policy = {
        "Version": "2008-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": "*",
            "Action": "s3:GetObject",
            "Resource": f'arn:aws:s3:::{bucket_name}/*',
            "Condition": {
                "StringEquals": {
                    "aws:sourceVpce": f'{vpc_endpoint}'
                }
            }
        }]
    }
    
    bucket_policy = json.dumps(bucket_policy)
    
    client = boto3.client('s3')
    client.put_bucket_policy(Bucket=bucket_name, Policy=bucket_policy)
    
def _setup_api_gateway_policy():
    print("Setup API Gateway Policy")
    
    api_policy = {
        "Version": "2012-10-17",
        "Statement": [{
                "Effect": "Allow",
                "Principal": "*",
                "Action": "execute-api:Invoke",
                "Resource": f'arn:aws:execute-api:ap-northeast-1:{account_id}:{api_gateway}/*',
                "Condition": {
                    "StringEquals": {
                        "aws:sourceVpce": [f'{vpc_endpoint}', f'{vpc_endpoint2}']
                    }
                }
        }]
    }
    
    api_policy = json.dumps(api_policy)

    client = boto3.client('apigateway')
    response = client.update_rest_api(
        restApiId=api_gateway_id,
        patchOperations=[
            {
                "op": "replace",
                'path': "/policy",
                'value': f'{api_policy}',
            },
        ]
    )

def handler(event, context):
    print("Setup Start")
    
    # S3バケットポリシー設定
    _setup_bucket_policy()
    
    # API Gatewayポリシー設定
    _setup_api_gateway_policy()
    
    print("Setup Done")

if __name__ == "__main__":
    handler(None, None)

処理内容はソースの下の方にある「handler」関数から読んでいったら分かりやすいかと思います。

S3バケットにポリシーを設定するため「_setup_bucket_policy」関数を呼び出しています。「_setup_bucket_policy」関数ではポリシーを定義したJSONを作成し、boto3のS3 Clientの「put_bucket_policy()」メソッドを呼びしてポリシーを設定しています。

次にAPI Gatewayにポリシーを設定するため「_setup_api_gateway_policy」関数を呼び出しています。「_setup_api_gateway_policy」関数でもポリシーを定義したJSONを作成しています。boto3のAPI Gateway Clientを使うのは先の「_setup_bucket_policy」と同じですが、ポリシーを設定するため「update_rest_api()」メソッドを呼び出しているところがS3とは異なっています。