JWT（JSON Web Token）認証を用いたAWS CDK（Python）によるCRUD APIの構築

# はじめに

**前回の記事**では、**AWS CDK（Python）**を活用して、**DynamoDB**、**Lambda**、および**API Gateway**を組み合わせた柔軟性の高いCRUD APIを構築し、**IPアドレス制限**を設定する方法について解説しました。
さらに、**APIキー認証**を導入する方法も紹介しました。

https://qiita.com/syukan3/items/8e8e808cedb22022534b

https://qiita.com/syukan3/items/f8ed0226a7e7a7f6bc81

今回は、より高度な認証手法として**JWT（JSON Web Token）認証**を導入する方法を詳しく説明します。
JWT認証は、ユーザー認証と認可をセキュアかつスケーラブルに管理するための強力な手段です。

---

## JWT認証とは

**JWT（JSON Web Token）認証**は、ユーザー認証と認可を行うためのオープン標準（RFC 7519）です。
JWTは、クレームと呼ばれる情報のセットを安全に伝達するためのコンパクトでURLセーフな方法を提供します。

**主な特徴**:

- **自己完結型**: ユーザー情報や認可情報をトークン内に含めるため、サーバー側でのセッション管理が不要。
- **スケーラブル**: 分散システムやマイクロサービスアーキテクチャに適している。
- **セキュア**: トークンは署名されており、改ざんを防止できる。必要に応じて暗号化も可能。

**注意点**:

- **トークンの管理**: トークンは機密情報として扱い、適切に保護する必要があります。
- **有効期限**: トークンには有効期限を設定し、定期的に更新することでセキュリティを維持します。

---

## プロジェクトのセットアップと前提条件

前回の記事で構築したプロジェクトを基に、JWT認証を導入します。
以下の前提条件が満たされていることを確認してください。

1. **前回の記事のセットアップ完了**: APIキー認証を設定したCRUD APIが既に構築・デプロイされていること。
2. **AWS CLIの設定**: 適切に設定されていること。
3. **AWS CDKのインストール**: 最新バージョンがインストールされていること。
4. **Python仮想環境のアクティブ化**: 前回と同様に設定されていること。
5. **AWS Cognitoの基本理解**: ユーザープールとアプリクライアントの基本的な概念を理解していること。

---

## AWS Cognitoの設定

JWT認証を実現するために、**AWS Cognito**を利用します。
Cognitoは、ユーザー管理、認証、認可を提供するマネージドサービスです。

### 1. Cognitoユーザープールの作成

Cognitoユーザープールは、ユーザーの登録やログインを管理するためのサービスです。

#### 手動での設定（AWSマネジメントコンソールを使用）

1. **AWSマネジメントコンソール**にログイン。
2. **Cognito**サービスに移動。
3. **Manage User Pools**を選択し、**Create a user pool**をクリック。
4. **Pool name**に「CRUDApiUserPool」と入力し、**Review defaults**を選択。
5. 必要に応じて設定を変更し、**Create pool**をクリック。

#### CDKを使用した設定（推奨）

CognitoユーザープールをCDKスタックに統合することで、インフラストラクチャをコードとして管理できます。
次節で詳しく説明します。

---

## CDKスタックの更新

JWT認証を導入するために、CDKスタックを以下のように更新します。
主な変更点は、Cognitoユーザープールの作成、API Gatewayに認証を適用することです。

### 1. Cognitoユーザープールの作成

`cdk_crud_api_python/cdk_crud_api_python_stack.py` を以下のように更新します。

```python
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)
```

**主な変更点**:

1. **Cognitoユーザープールの作成**:
    ```python
    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」に設定。
    - ユーザーの自己登録を有効化。
    - ユーザー名とメールアドレスによるサインインを許可。
    - メールアドレスの自動検証を有効化。

2. **Cognitoアプリクライアントの作成**:
    ```python
    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）フローを有効化。

3. **Cognitoオーソライザーの作成**:
    ```python
    authorizer = apigateway.CognitoUserPoolsAuthorizer(
        self, "CognitoAuthorizer",
        cognito_user_pools=[user_pool]
    )
    ```
    - API Gateway用のCognitoオーソライザーを作成し、ユーザープールを指定。

4. **APIメソッドに認証の適用**:
    ```python
    items.add_method("GET", apigateway.LambdaIntegration(crud_lambda), authorization_type=apigateway.AuthorizationType.COGNITO, authorizer=authorizer)
    ```
    - 各APIメソッドに対して、`authorization_type`を`COGNITO`に設定し、先ほど作成したオーソライザーを適用。

---

### 2. API Gatewayに認証を適用

CognitoオーソライザーをAPI Gatewayの各メソッドに適用することで、JWTトークンを使用した認証が可能になります。
上記のコード例では、`authorization_type`と`authorizer`を設定することで実現しています。

---

## Lambda関数の更新

今回の変更では、Lambda関数側でJWT認証を直接処理する必要はありません。
認証はAPI Gatewayが行い、認証済みのリクエストのみがLambda関数に渡されます。
ただし、必要に応じてLambda関数内でユーザー情報を取得・利用することができます。

### 1. Lambdaディレクトリの作成

前回の記事で既に`lambda`フォルダを作成している場合は、このステップは不要です。
まだ作成していない場合は、プロジェクトのルートディレクトリに`lambda`フォルダを作成し、その中に`index.py`ファイルを作成します。

```bash
mkdir lambda
touch lambda/index.py
```

### 2. Lambdaハンドラーの実装

`lambda/index.py`に以下のコードを追加します。
これは、DynamoDBテーブルへのCRUD操作を実装したLambdaハンドラーの完全な例です。
JWT認証はAPI Gatewayが処理するため、Lambda関数側では特別な認証処理は不要です。
ただし、ユーザー情報を取得する場合は、イベントオブジェクトからユーザー情報を参照できます。

```python
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ユーザー情報の取得**:
    ```python
    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`などのクレームを取得して、作成者や更新者の情報として利用できます。

- **CRUD操作の実装**:
    - 各HTTPメソッドとリソースパスに応じて、DynamoDBテーブルに対する操作を行います。
    - **POST /items**では、作成者情報を追加。
    - **PUT /items/{id}**では、更新者情報を追加。

- **エラーハンドリング**:
    ```python
    except Exception as e:
        print(e)
        return {
            "statusCode": 500,
            "body": json.dumps({"message": "Internal Server Error"})
        }
    ```
    - 例外が発生した場合、`500 Internal Server Error`を返します。デバッグのためにエラーをログ出力しています。

---

## デプロイとテスト

### 1. CDKスタックのデプロイ

更新したCDKスタックをデプロイします。

```bash
cdk deploy
```

デプロイが成功すると、API GatewayのエンドポイントURLとCognitoユーザープールの情報が表示されます。

### 2. Cognitoユーザーの登録とトークンの取得

#### 1. ユーザーの登録

ユーザープールにユーザーを登録します。
AWSマネジメントコンソールを使用するか、AWS CLIを使用して登録できます。

**AWSマネジメントコンソールを使用する場合**:

1. **AWSマネジメントコンソール**にログイン。
2. **Cognito**サービスに移動。
3. 作成した**CRUDApiUserPool**を選択。
4. **Users and groups**タブを選択し、**Create user**をクリック。
5. 必要な情報（ユーザー名、メールアドレスなど）を入力し、**Create user**をクリック。

#### 2. ClientIdの取得

`YOUR_USER_POOL_ID`はAWSマネジメントコンソールから取得してください。

```sh
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`先ほど登録したユーザー情報を入力してください。

```bash
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
```

出力例:

```json
{
    "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`)

```bash
curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items \
     -H "Authorization: Bearer YOUR_ID_TOKEN"
```

#### 2. 新しいアイテムの作成 (`POST /items`)

```bash
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}`)

```bash
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}`)

```bash
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}`)

```bash
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 ドキュメント](https://docs.aws.amazon.com/cdk/latest/guide/home.html)
- [AWS CDK API Reference (Python)](https://docs.aws.amazon.com/cdk/api/v2/python/index.html)
- [AWS Lambda ドキュメント](https://docs.aws.amazon.com/lambda/latest/dg/welcome.html)
- [Amazon DynamoDB ドキュメント](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Introduction.html)
- [Amazon API Gateway ドキュメント](https://docs.aws.amazon.com/apigateway/latest/developerguide/welcome.html)
- [AWS Cognito ドキュメント](https://docs.aws.amazon.com/cognito/latest/developerguide/what-is-amazon-cognito.html)
- [Python dotenv ドキュメント](https://pypi.org/project/python-dotenv/)
- [JWT.io](https://jwt.io/)

---

**注記**: セキュリティの観点から、`.env`ファイルには機密情報を含めないようにし、必要に応じてAWS Secrets Managerなどのサービスを利用して機密情報を管理してください。また、JWTトークンは適切に管理し、不正アクセスを防止するために定期的にトークンの有効期限を設定・更新することを推奨します。

---
### 会社紹介

**株式会社 Mosaica**  
最先端テクノロジーで社会課題を解決し、持続可能な未来を創造する IT カンパニー。
AI ソリューション、クラウド統合、DX 推進、経営コンサルティングなど包括的なサービスでビジネス変革を支援しています。  

詳しくは [公式サイト](https://mosaica.co.jp/) までお気軽にご相談ください。
公式サイト： https://mosaica.co.jp/