【Python】CDKでAPIGateway+Lambda認証をしてみる
何番煎じですかね？
今回はAPI Gateway + lambdaをcdkで実装してみますーというのをゼロベースで作っていきますー
今回作るものはこんな感じ
では 作っていきましょう👻
プロジェクトを用意する
雛形を用意
$ mkdir cdk_api_gateway_lambda_authorizer
$ cd cdk_api_gateway_lambda_authorizer
$ cdk init --language python
$ rm setup.py source.bat
# Lambdaのディレクトリも作っておく
$ mkdir lambda
依存パッケージ
aws-cdk.core==1.121.0
aws-cdk.aws-lambda==1.121.0
aws_cdk.aws_lambda_python==1.121.0
aws-cdk.aws_apigateway==1.121.0
インストール
$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install -r requirements.txt
Stackを用意する
from aws_cdk import core as cdk
from aws_cdk.aws_apigateway import IdentitySource, LambdaIntegration, RestApi, TokenAuthorizer
from aws_cdk.aws_lambda import Runtime
from aws_cdk.aws_lambda_python import PythonFunction
APP_NAME = "CdkApiGatewayLambdaAuthorizer"
class CdkApiGatewayLambdaAuthorizerStack(cdk.Stack):
def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
common_lambda_param = {"entry": "lambda", "runtime": Runtime.PYTHON_3_8}
auth_lambda = PythonFunction(self, f"{APP_NAME}AuthLambda", handler="authorizer_handler", **common_lambda_param)
hello_lambda = PythonFunction(self, f"{APP_NAME}HelloLambda", handler="hello_handler", **common_lambda_param)
api = RestApi(self, f"{APP_NAME}Api")
auth = TokenAuthorizer(
self, "hoge_authorizer", identity_source=IdentitySource.header("HogeAuthorization"), handler=auth_lambda
)
api.root.add_method("GET", LambdaIntegration(hello_lambda), authorizer=auth)
lambdaをauth_lambdaとhello_lambdaで2つ用意してます。
それぞれ認証用とHelloのAPI動作確認用です。
API Gateway はAPIタイプをRestAPIで作成して、
auth_lambdaのトークン認証でルートにGETメソッドにhello_lambdaを設定している感じです。
アプリを書く
まずHello側から、、、
def hello(authorizer_context):
return {
"statusCode": 200,
"body": f"Hello world! authorizer_context = {authorizer_context}",
}
完全に動作確認用なので特に説明なしです。
次に認証。。
def authorizer(token, resources):
if (length := len(token)) <= 1: # 動作確認向けのへぼへぼ認証
raise Exception("Unauthorized")
return {
"principalId": "hogehoge_principal_id",
"policyDocument": {
"Version": "2012-10-17",
"Statement": [{"Action": "execute-api:Invoke", "Effect": "Allow", "Resource": resources}],
},
"context": {"hogehoge_token_length": length},
}
動きは、tokenをもらって、その長さが2文字以上なら認証成功！！ それ以下なら失敗という動作確認のための認証機能を持ってます。
トークン認証では、返り値にprincipalIdとpolicyDocumentが必要です。
contextはなくても良いですが、principalId以外に後ろのLambdaに渡したい値があればKeyValueで指定できます。
principalIdはユーザーID的なやつです。このリクエストが誰からのリクエストなのかを後ろに伝える用途ですが、今回はsampleで固定値いれちゃってます。
policyDocumentはこのリクエストのポリシーです。これはリクエスト毎に作らなければいけないので、やや面倒なやつです。
エレメントとしてはVersionとStatementがあれば大丈夫です。
Versionは固定値なのでStatementですが、大体こんな感じです。
{
"Action": "execute-api:Invoke", // ←API GatewayがLambdaの実行を...
"Effect": "Allow", // 許可します！
"Resource": hogehogeho // リソースははhogehoge
}`
これ以外にもIPアドレスによる制限とかもできますが、このへんはIAMの話なので割愛です。
ここでResourceのARNですが「一体どうやって指定すんのよ？」と思うかもしれません。
リクエストで叩かれたエンドポイントのARNは、eventに入ってハンドラーに渡されるのでそれを使えばOKです。
はい。。
話を戻して、最後にindexです
from authorizer import authorizer
from hello import hello
def authorizer_handler(event, context):
token = event["authorizationToken"] # リクエストヘッダーの`HogeAuthorization`で指定した値が入る
resources = [event["methodArn"]] # 叩かれたエンドポイントのリソースが入る
return authorizer(token, resources)
def hello_handler(event, context):
authorizer_context = event["requestContext"]["authorizer"] # 認証lambdaから渡された`context`とか`principalId`とかが入る
return hello(authorizer_context)
ハンドラーに紐付ける関数をimportして、eventから必要な値をとってきて流してるだけです。
デプロイ
下記のコマンドでデプロイ！
$ cdk deploy
動作確認
デプロイが成功するとエンドポイントが表示されるので、そこに対してcurlで動作確認をしてみます。
# HogeAuthorizationが2文字未満だと失敗する。
$ curl https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod --header 'HogeAuthorization: a'
> {"message":"Unauthorized"}
# HogeAuthorizationが2文字以上だと成功する
$ curl https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod --header 'HogeAuthorization: aa'
> Hello world! authorizer_context = {'principalId': 'hogehoge_principal_id', 'integrationLatency': 251, 'hogehoge_token_length': '2'}
以上〜！
今回のリポジトリはこちら
https://github.com/sisi100/cdk_api_gateway_lambda_authorizer