はじめに
前回の記事では、AWS CDK（Python）を活用して、DynamoDB、Lambda、およびAPI Gatewayを組み合わせた柔軟性の高いCRUD APIを構築し、IPアドレス制限を設定する方法について解説しました。
さらに、APIキー認証を導入する方法も紹介しました。
今回は、より高度な認証手法としてJWT（JSON Web Token）認証を導入する方法を詳しく説明します。
JWT認証は、ユーザー認証と認可をセキュアかつスケーラブルに管理するための強力な手段です。
JWT認証とは
JWT（JSON Web Token）認証は、ユーザー認証と認可を行うためのオープン標準（RFC 7519）です。
JWTは、クレームと呼ばれる情報のセットを安全に伝達するためのコンパクトでURLセーフな方法を提供します。
主な特徴:
- 自己完結型: ユーザー情報や認可情報をトークン内に含めるため、サーバー側でのセッション管理が不要。
- スケーラブル: 分散システムやマイクロサービスアーキテクチャに適している。
- セキュア: トークンは署名されており、改ざんを防止できる。必要に応じて暗号化も可能。
注意点:
- トークンの管理: トークンは機密情報として扱い、適切に保護する必要があります。
- 有効期限: トークンには有効期限を設定し、定期的に更新することでセキュリティを維持します。
プロジェクトのセットアップと前提条件
前回の記事で構築したプロジェクトを基に、JWT認証を導入します。
以下の前提条件が満たされていることを確認してください。
- 前回の記事のセットアップ完了: APIキー認証を設定したCRUD APIが既に構築・デプロイされていること。
- AWS CLIの設定: 適切に設定されていること。
- AWS CDKのインストール: 最新バージョンがインストールされていること。
- Python仮想環境のアクティブ化: 前回と同様に設定されていること。
- AWS Cognitoの基本理解: ユーザープールとアプリクライアントの基本的な概念を理解していること。
AWS Cognitoの設定
JWT認証を実現するために、AWS Cognitoを利用します。
Cognitoは、ユーザー管理、認証、認可を提供するマネージドサービスです。
1. Cognitoユーザープールの作成
Cognitoユーザープールは、ユーザーの登録やログインを管理するためのサービスです。
手動での設定（AWSマネジメントコンソールを使用）
- AWSマネジメントコンソールにログイン。
- Cognitoサービスに移動。
- Manage User Poolsを選択し、Create a user poolをクリック。
- Pool nameに「CRUDApiUserPool」と入力し、Review defaultsを選択。
- 必要に応じて設定を変更し、Create poolをクリック。
CDKを使用した設定（推奨）
CognitoユーザープールをCDKスタックに統合することで、インフラストラクチャをコードとして管理できます。
次節で詳しく説明します。
CDKスタックの更新
JWT認証を導入するために、CDKスタックを以下のように更新します。
主な変更点は、Cognitoユーザープールの作成、API Gatewayに認証を適用することです。
1. Cognitoユーザープールの作成
cdk_crud_api_python/cdk_crud_api_python_stack.py を以下のように更新します。
from aws_cdk import (
Stack,
aws_dynamodb as dynamodb,
aws_lambda as _lambda,
aws_apigateway as apigateway,
aws_cognito as cognito,
aws_iam as iam,
RemovalPolicy
)
from constructs import Construct
import os
from dotenv import load_dotenv
load_dotenv()
class CdkCrudApiPythonStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 環境変数からDynamoDBのキー設定を取得
partition_key = os.getenv("DYNAMODB_PARTITION_KEY", "id")
sort_key = os.getenv("DYNAMODB_SORT_KEY") # 必要に応じて設定
# DynamoDBテーブルの作成
table = dynamodb.Table(
self, "ItemsTable",
partition_key=dynamodb.Attribute(
name=partition_key,
type=dynamodb.AttributeType.STRING
),
sort_key=dynamodb.Attribute(name=sort_key, type=dynamodb.AttributeType.STRING) if sort_key else None,
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
removal_policy=RemovalPolicy.DESTROY # 開発環境用。プロダクションでは注意
)
# Cognitoユーザープールの作成
user_pool = cognito.UserPool(
self, "UserPool",
user_pool_name="CRUDApiUserPool",
self_sign_up_enabled=True,
sign_in_aliases=cognito.SignInAliases(username=True, email=True),
auto_verify=cognito.AutoVerifiedAttrs(email=True),
standard_attributes=cognito.StandardAttributes(
email=cognito.StandardAttribute(required=True, mutable=False)
)
)
# Cognitoアプリクライアントの作成
user_pool_client = cognito.UserPoolClient(
self, "UserPoolClient",
user_pool=user_pool,
generate_secret=False,
auth_flows=cognito.AuthFlow(
user_password=True,
user_srp=True
)
)
# Lambda関数の作成
crud_lambda = _lambda.Function(
self, "CrudFunction",
runtime=_lambda.Runtime.PYTHON_3_9,
handler="index.handler",
code=_lambda.Code.from_asset("lambda"),
environment={
"TABLE_NAME": table.table_name
}
)
# LambdaにDynamoDBへのアクセス権限を付与
table.grant_read_write_data(crud_lambda)
# API Gatewayの作成
api = apigateway.RestApi(
self, "CrudApi",
rest_api_name="CRUD Service",
description="This service serves CRUD operations.",
default_cors_preflight_options=apigateway.CorsOptions(
allow_origins=apigateway.Cors.ALL_ORIGINS,
allow_methods=apigateway.Cors.ALL_METHODS
)
)
# Cognitoオーソライザーの作成
authorizer = apigateway.CognitoUserPoolsAuthorizer(
self, "CognitoAuthorizer",
cognito_user_pools=[user_pool]
)
# APIリソースとメソッドの設定
items = api.root.add_resource("items")
items.add_method("GET", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
items.add_method("POST", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
single_item = items.add_resource("{id}")
single_item.add_method("GET", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
single_item.add_method("PUT", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
single_item.add_method("DELETE", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
主な変更点:
-
Cognitoユーザープールの作成:
user_pool = cognito.UserPool( self, "UserPool", user_pool_name="CRUDApiUserPool", self_sign_up_enabled=True, sign_in_aliases=cognito.SignInAliases(username=True, email=True), auto_verify=cognito.AutoVerifiedAttrs(email=True), standard_attributes=cognito.StandardAttributes( email=cognito.StandardAttribute(required=True, mutable=False) ) )- ユーザープール名を「CRUDApiUserPool」に設定。
- ユーザーの自己登録を有効化。
- ユーザー名とメールアドレスによるサインインを許可。
- メールアドレスの自動検証を有効化。
-
Cognitoアプリクライアントの作成:
user_pool_client = cognito.UserPoolClient( self, "UserPoolClient", user_pool=user_pool, generate_secret=False, auth_flows=cognito.AuthFlow( user_password=True, user_srp=True ) )- アプリクライアントを作成し、秘密鍵の生成を無効化。
- ユーザーパスワードとSRP（Secure Remote Password）フローを有効化。
-
Cognitoオーソライザーの作成:
authorizer = apigateway.CognitoUserPoolsAuthorizer( self, "CognitoAuthorizer", cognito_user_pools=[user_pool] )- API Gateway用のCognitoオーソライザーを作成し、ユーザープールを指定。
-
APIメソッドに認証の適用:
items.add_method("GET", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)- 各APIメソッドに対して、
authorization_typeをCOGNITOに設定し、先ほど作成したオーソライザーを適用。
- 各APIメソッドに対して、
2. API Gatewayに認証を適用
CognitoオーソライザーをAPI Gatewayの各メソッドに適用することで、JWTトークンを使用した認証が可能になります。
上記のコード例では、authorization_typeとauthorizerを設定することで実現しています。
Lambda関数の更新
今回の変更では、Lambda関数側でJWT認証を直接処理する必要はありません。
認証はAPI Gatewayが行い、認証済みのリクエストのみがLambda関数に渡されます。
ただし、必要に応じてLambda関数内でユーザー情報を取得・利用することができます。
1. Lambdaディレクトリの作成
前回の記事で既にlambdaフォルダを作成している場合は、このステップは不要です。
まだ作成していない場合は、プロジェクトのルートディレクトリにlambdaフォルダを作成し、その中にindex.pyファイルを作成します。
mkdir lambda
touch lambda/index.py
2. Lambdaハンドラーの実装
lambda/index.pyに以下のコードを追加します。
これは、DynamoDBテーブルへのCRUD操作を実装したLambdaハンドラーの完全な例です。
JWT認証はAPI Gatewayが処理するため、Lambda関数側では特別な認証処理は不要です。
ただし、ユーザー情報を取得する場合は、イベントオブジェクトからユーザー情報を参照できます。
import json
import os
import boto3
import uuid
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])
def handler(event, context):
http_method = event['httpMethod']
resource = event['resource']
path_parameters = event.get('pathParameters', {})
item_id = path_parameters.get('id') if path_parameters else None
# Cognitoユーザー情報の取得（オプション）
user_info = get_user_info(event)
try:
if resource == "/items" and http_method == "GET":
# 全アイテムの取得
response = table.scan()
return {
"statusCode": 200,
"body": json.dumps(response.get('Items', []))
}
elif resource == "/items" and http_method == "POST":
# 新しいアイテムの作成
body = json.loads(event['body'])
if 'id' not in body:
body['id'] = generate_id()
# 作成者情報の追加（オプション）
body['created_by'] = user_info.get('username') if user_info else "anonymous"
table.put_item(Item=body)
return {
"statusCode": 201,
"body": json.dumps(body)
}
elif resource == "/items/{id}" and http_method == "GET":
# 単一アイテムの取得
response = table.get_item(Key={'id': item_id})
item = response.get('Item')
if item:
return {
"statusCode": 200,
"body": json.dumps(item)
}
else:
return {
"statusCode": 404,
"body": json.dumps({"message": "Item not found"})
}
elif resource == "/items/{id}" and http_method == "PUT":
# アイテムの更新
body = json.loads(event['body'])
body['id'] = item_id
# 更新者情報の追加（オプション）
body['updated_by'] = user_info.get('username') if user_info else "anonymous"
table.put_item(Item=body)
return {
"statusCode": 200,
"body": json.dumps(body)
}
elif resource == "/items/{id}" and http_method == "DELETE":
# アイテムの削除
table.delete_item(Key={'id': item_id})
return {
"statusCode": 204,
"body": ""
}
else:
return {
"statusCode": 405,
"body": json.dumps({"message": "Method Not Allowed"})
}
except Exception as e:
print(e)
return {
"statusCode": 500,
"body": json.dumps({"message": "Internal Server Error"})
}
def generate_id():
return str(uuid.uuid4())
def get_user_info(event):
"""
Cognitoユーザー情報を取得します。
API GatewayのCognitoオーソライザーを使用している場合、ユーザー情報はeventオブジェクトに含まれています。
"""
try:
authorizer = event['requestContext']['authorizer']
claims = authorizer['claims']
return {
"username": claims.get('cognito:username'),
"email": claims.get('email')
}
except KeyError:
return None
ポイント解説:
-
Cognitoユーザー情報の取得:
def get_user_info(event): try: authorizer = event['requestContext']['authorizer'] claims = authorizer['claims'] return { "username": claims.get('cognito:username'), "email": claims.get('email') } except KeyError: return None- API GatewayのCognitoオーソライザーを使用している場合、
event['requestContext']['authorizer']['claims']にユーザー情報が含まれています。 -
cognito:usernameやemailなどのクレームを取得して、作成者や更新者の情報として利用できます。
- API GatewayのCognitoオーソライザーを使用している場合、
-
CRUD操作の実装:
- 各HTTPメソッドとリソースパスに応じて、DynamoDBテーブルに対する操作を行います。
- POST /itemsでは、作成者情報を追加。
- **PUT /items/{id}**では、更新者情報を追加。
-
エラーハンドリング:
except Exception as e: print(e) return { "statusCode": 500, "body": json.dumps({"message": "Internal Server Error"}) }- 例外が発生した場合、
500 Internal Server Errorを返します。デバッグのためにエラーをログ出力しています。
- 例外が発生した場合、
デプロイとテスト
1. CDKスタックのデプロイ
更新したCDKスタックをデプロイします。
cdk deploy
デプロイが成功すると、API GatewayのエンドポイントURLとCognitoユーザープールの情報が表示されます。
2. Cognitoユーザーの登録とトークンの取得
1. ユーザーの登録
ユーザープールにユーザーを登録します。
AWSマネジメントコンソールを使用するか、AWS CLIを使用して登録できます。
AWSマネジメントコンソールを使用する場合:
- AWSマネジメントコンソールにログイン。
- Cognitoサービスに移動。
- 作成したCRUDApiUserPoolを選択。
- Users and groupsタブを選択し、Create userをクリック。
- 必要な情報（ユーザー名、メールアドレスなど）を入力し、Create userをクリック。
2. ClientIdの取得
YOUR_USER_POOL_IDはAWSマネジメントコンソールから取得してください。
aws cognito-idp list-user-pool-clients --user-pool-id YOUR_USER_POOL_ID
{
"UserPoolClients": [
{
"ClientId": "6sd975gago...",
"UserPoolId": "ap-northeast-1_...",
"ClientName": "UserPoolClient..."
}
]
}
3. JWTトークンの取得
ユーザー登録後、JWTトークン（IDトークンとアクセストークン）を取得します。
以下は、AWS CLIを使用してトークンを取得する例です。
USER_PASSWORD_AUTHはそのままにしてください。
YOUR_USER_POOL_CLIENT_IDは先ほど取得したClientIdに置き換えてください。
your_username、your_password先ほど登録したユーザー情報を入力してください。
aws cognito-idp initiate-auth \
--auth-flow USER_PASSWORD_AUTH \
--client-id YOUR_USER_POOL_CLIENT_ID \
--auth-parameters USERNAME=your_username,PASSWORD=your_password
出力例:
{
"AuthenticationResult": {
"AccessToken": "eyJraWQiOiJr...",
"ExpiresIn": 3600,
"IdToken": "eyJraWQiOiJr...",
"RefreshToken": "eyJjdHkiOiJKV1QiLCJ...",
"TokenType": "Bearer"
}
}
- AccessToken: 認可情報を含むトークン。API Gatewayでの認証に使用します。
- IdToken: ユーザー情報を含むトークン。Lambda関数でユーザー情報を取得する際に使用します。
- RefreshToken: トークンの再取得に使用します。
（4. 初回ログイン時にパスワード変更が必要な場合）
初回ログイン時にパスワード変更が必要な場合があります。
その場合は、以下のコマンドを実行すると、AccessTokenなどを取得することができます。
UserPoolIdはユーザープールIDに置き換えてください。
YOUR_USER_POOL_CLIENT_IDは先ほど取得したClientIDに置き換えてください。
NEW_PASSWORD_REQUIREDはそのままにしてください。
your_usernameは作成したユーザー名で置き換えてください。
your_passwordは新しいパスワードに置き換えてください。
SESSIONはJWTトークンの取得で返ってきたセッションに置き換えてください。
aws cognito-idp admin-respond-to-auth-challenge \
--user-pool-id UserPoolId \
--client-id YOUR_USER_POOL_CLIENT_ID \
--challenge-name NEW_PASSWORD_REQUIRED \
--challenge-responses "USERNAME=your_username,NEW_PASSWORD=your_password" \
--session "SESSION"
4. APIのテスト方法
JWTトークンを使用してAPIにアクセスする際は、リクエストヘッダーにAuthorizationを含めます。
以下にcurlコマンドを使用したテスト方法を示します。
1. 全アイテムの取得 (GET /items)
curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items \
-H "Authorization: Bearer YOUR_ID_TOKEN"
2. 新しいアイテムの作成 (POST /items)
curl -X POST https://your-api-id.execute-api.region.amazonaws.com/prod/items \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ID_TOKEN" \
-d '{"name": "Item Name", "description": "Item Description"}'
3. 単一アイテムの取得 (GET /items/{id})
curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id> \
-H "Authorization: Bearer YOUR_ID_TOKEN"
4. アイテムの更新 (PUT /items/{id})
curl -X PUT https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id> \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ID_TOKEN" \
-d '{"name": "Updated Name", "description": "Updated Description"}'
5. アイテムの削除 (DELETE /items/{id})
curl -X DELETE https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id> \
-H "Authorization: Bearer YOUR_ID_TOKEN"
JWTトークンが無効または提供されていない場合:
リクエストは401 Unauthorizedエラーを返します。適切なJWTトークンをヘッダーに含める必要があります。
最終的なディレクトリ構成
プロジェクト全体の最終的なディレクトリ構成は以下の通りです。
cdk-crud-api-python/
├── app.py
├── cdk_crud_api_python
│ ├── __init__.py
│ └── cdk_crud_api_python_stack.py
├── lambda
│ └── index.py
├── requirements.txt
├── .gitignore
├── README.md
├── cdk.json
└── .env
各ファイル・フォルダの役割:
- app.py: CDKアプリケーションのエントリーポイント。
-
cdk_crud_api_python/: CDKスタックの定義が含まれるパッケージ。
- init.py: パッケージ初期化ファイル。
- cdk_crud_api_python_stack.py: DynamoDB、Lambda、API Gateway、およびCognitoを定義するスタックファイル。
-
lambda/: Lambda関数のコードが含まれるディレクトリ。
- index.py: CRUDロジックを実装したLambdaハンドラー。
- requirements.txt: Pythonの依存関係リスト。
- .gitignore: Gitで追跡しないファイル・フォルダを指定。
- README.md: プロジェクトの説明や使用方法を記載。
- cdk.json: CDKの設定ファイル。
- .env: 環境変数を定義するファイル（セキュリティに注意）。
まとめ
今回のガイドでは、AWS CDK（Python）を使用して構築したCRUD APIにJWT（JSON Web Token）認証を導入する方法を詳しく解説しました。
Cognitoユーザープールを活用することで、ユーザー管理と認証を簡単に実現し、セキュアなAPIアクセスを提供できます。
ポイントまとめ:
- Cognitoユーザープールの作成と設定: ユーザー登録、認証フローを設定し、アプリクライアントを作成。
- API Gatewayへの認証適用: Cognitoオーソライザーを使用して、APIメソッドに認証を適用。
- Lambda関数の柔軟な対応: Lambda関数内でユーザー情報を取得・利用可能。
- セキュリティの向上: JWTトークンを利用することで、セッション管理を不要にし、スケーラブルな認証を実現。
参考資料
- AWS CDK ドキュメント
- AWS CDK API Reference (Python)
- AWS Lambda ドキュメント
- Amazon DynamoDB ドキュメント
- Amazon API Gateway ドキュメント
- AWS Cognito ドキュメント
- Python dotenv ドキュメント
- JWT.io
注記: セキュリティの観点から、.envファイルには機密情報を含めないようにし、必要に応じてAWS Secrets Managerなどのサービスを利用して機密情報を管理してください。また、JWTトークンは適切に管理し、不正アクセスを防止するために定期的にトークンの有効期限を設定・更新することを推奨します。