はじめに

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

今回の記事では、IP制限に代わる認証手法としてAPIキー認証を導入する方法を詳しく説明します。
APIキー認証は、特定のユーザーやアプリケーションに対してアクセス権を管理するためのシンプルかつ効果的な方法です。

APIキー認証とは

APIキー認証は、クライアントがAPIにアクセスする際に一意のキーを提供することで認証を行うシンプルな方法です。
APIキーは通常、リクエストのヘッダーやクエリパラメータに含められ、API Gatewayはこのキーを検証してリクエストを許可または拒否します。

主な特徴:

シンプル: 実装が容易で、迅速に導入可能。
制御可能: キーごとにアクセス制限や使用量の制御が可能。
管理可能: APIキーの発行、無効化、ローテーションが容易。

注意点:

セキュリティ: APIキーは容易に共有・漏洩する可能性があるため、機密情報として扱う必要があります。機密性の高いデータや操作には、より強力な認証手段（例：OAuth2.0）を検討してください。

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

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

前回の記事のセットアップ完了: IP制限を設定したCRUD APIが既に構築・デプロイされていること。
AWS CLIの設定: 適切に設定されていること。
AWS CDKのインストール: 最新バージョンがインストールされていること。
Python仮想環境のアクティブ化: 前回と同様に設定されていること。

CDKスタックの更新

APIキー認証を導入するために、CDKスタックを以下のように更新します。
主な変更点は、APIキーと使用プランの作成、API Gatewayの設定変更です。

1. APIキーと使用プランの作成

まず、APIキーと使用プラン（Usage Plan）を作成し、API Gatewayに関連付けます。
これにより、特定のAPIキーを持つクライアントのみがAPIにアクセスできるようになります。

以下のように、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,
    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 # 開発環境用。プロダクションでは注意
        )

        # 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
            )
        )

        # APIキーと使用プランの作成
        api_key = api.add_api_key("ApiKey")

        usage_plan = api.add_usage_plan(
            "UsagePlan",
            name="BasicUsagePlan",
            throttle=apigateway.ThrottleSettings(
                rate_limit=100,  # リクエスト数の上限（1秒あたり）
                burst_limit=200  # バースト許容数
            )
        )

        usage_plan.add_api_stage(
            stage=api.deployment_stage
        )

        # APIリソースとメソッドの設定
        items = api.root.add_resource("items")
        get_items = items.add_method("GET", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
        post_items = items.add_method("POST", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)

        single_item = items.add_resource("{id}")
        get_single_item = single_item.add_method("GET", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
        put_single_item = single_item.add_method("PUT", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
        delete_single_item = single_item.add_method("DELETE", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)

        # 使用プランにAPIステージを関連付け
        usage_plan.add_api_stage(
            stage=api.deployment_stage,
            throttle=[
                apigateway.ThrottleSettings(
                    method=get_items,
                    throttle=apigateway.ThrottleSettings(
                        rate_limit=50,
                        burst_limit=100
                    )
                ),
                apigateway.ThrottleSettings(
                    method=post_items,
                    throttle=apigateway.ThrottleSettings(
                        rate_limit=50,
                        burst_limit=100
                    )
                )
                # 必要に応じて他のメソッドも追加
            ]
        )

主な変更点:

APIキーの作成:
```
api_key = api.add_api_key("ApiKey")
```
add_api_keyメソッドを使用して、新しいAPIキーを作成します。これにより、AWSが自動的にAPIキーを生成します。
使用プラン（Usage Plan）の作成:
```
usage_plan = api.add_usage_plan(
    "UsagePlan",
    name="BasicUsagePlan",
    throttle=apigateway.ThrottleSettings(
        rate_limit=100,
        burst_limit=200
    )
)
```
使用プランは、APIキーごとのリクエスト制限やスロットリングを管理します。ここでは、1秒あたり100リクエストのレートリミットと200のバースト許容数を設定しています。
使用プランへのAPIステージの追加:
```
usage_plan.add_api_stage(
    stage=api.deployment_stage
)
```
使用プランをAPIのステージに関連付けます。

メソッドごとのAPIキーの要求設定:

get_items = items.add_method("GET", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
post_items = items.add_method("POST", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)

single_item = items.add_resource("{id}")
get_single_item = single_item.add_method("GET", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
put_single_item = single_item.add_method("PUT", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)
delete_single_item = single_item.add_method("DELETE", apigateway.LambdaIntegration(crud_lambda), api_key_required=True)

各APIメソッドに対して、api_key_required=Trueを設定することで、APIキーの提供が必須となります。

使用プランにメソッドごとのスロットリング設定を追加:

usage_plan.add_api_stage(
    stage=api.deployment_stage,
    throttle=[
        apigateway.ThrottleSettings(
            method=get_items,
            throttle=apigateway.ThrottleSettings(
                rate_limit=50,
                burst_limit=100
            )
        ),
        apigateway.ThrottleSettings(
            method=post_items,
            throttle=apigateway.ThrottleSettings(
                rate_limit=50,
                burst_limit=100
            )
        )
        # 必要に応じて他のメソッドも追加
    ]
)

メソッドごとに細かいスロットリング設定を行い、特定のメソッドに対するリクエスト制限を強化します。

2. API Gatewayの設定変更

前述の通り、APIキーと使用プランを作成し、APIメソッドにAPIキーの必須設定を追加しました。これにより、クライアントは有効なAPIキーを提供することでAPIにアクセスできるようになります。

Lambda関数の実装

Lambda関数は、DynamoDBテーブルへのCRUD操作を行います。今回の変更では、APIキー認証をAPI Gatewayで管理するため、Lambda関数側では特別な認証処理は必要ありません。ただし、必要に応じてAPIキーの情報をLambda関数内で利用することも可能です。

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

プロジェクトのルートディレクトリにlambdaフォルダを作成し、その中にindex.pyファイルを作成します。

mkdir lambda
touch lambda/index.py

2. Lambdaハンドラーの実装

lambda/index.pyに以下のコードを追加します。これは、DynamoDBテーブルへのCRUD操作を実装した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

    # 必要に応じてAPIキーの取得（オプション）
    # api_key = event['headers'].get('x-api-key')
    # print(f"API Key: {api_key}")  # デバッグ用。プロダクションでは不要

    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()
            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
            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())

ポイント解説:

DynamoDBの初期化:
```
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])
```
環境変数から取得したテーブル名を使用してDynamoDBリソースを初期化します。
APIキーの取得（オプション）:
```
# api_key = event['headers'].get('x-api-key')
# print(f"API Key: {api_key}")  # デバッグ用。プロダクションでは不要
```
必要に応じて、リクエストヘッダーからAPIキーを取得することができます。デバッグ目的でのログ出力は可能ですが、プロダクション環境では不要です。
CRUD操作の実装:
各HTTPメソッドとリソースパスに応じて、DynamoDBテーブルに対する操作を行います。
- GET /items: 全アイテムの取得
- POST /items: 新しいアイテムの作成
- GET /items/{id}: 単一アイテムの取得
- PUT /items/{id}: アイテムの更新
- DELETE /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が表示されます。

2. APIキーの取得と管理

デプロイ後、AWSマネジメントコンソールからAPIキーを取得します。

AWSマネジメントコンソールにログイン。
API Gatewayサービスに移動。
作成したCRUD APIを選択。
左側のメニューからAPIキーを選択。
作成されたAPIキーを確認し、必要に応じてコピーします。

APIキーの配布:
クライアントにAPIキーを安全に配布します。メールやセキュアなチャネルを利用して共有してください。

3. APIのテスト方法

APIキーを使用してAPIにアクセスする際は、リクエストヘッダーにx-api-keyを含めます。以下にcurlコマンドを使用したテスト方法を示します。

1. 全アイテムの取得 (`GET /items`)

curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items \
     -H "x-api-key: YOUR_API_KEY"

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

curl -X POST https://your-api-id.execute-api.region.amazonaws.com/prod/items \
     -H "Content-Type: application/json" \
     -H "x-api-key: YOUR_API_KEY" \
     -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 "x-api-key: YOUR_API_KEY"

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 "x-api-key: YOUR_API_KEY" \
     -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 "x-api-key: YOUR_API_KEY"

APIキーが無効または提供されていない場合:
リクエストは403 Forbiddenエラーを返します。適切なAPIキーをヘッダーに含める必要があります。

最終的なディレクトリ構成

プロジェクト全体の最終的なディレクトリ構成は以下の通りです。

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、およびAPIキー認証を定義するスタックファイル。
lambda/: Lambda関数のコードが含まれるディレクトリ。
- index.py: CRUDロジックを実装したLambdaハンドラー。
requirements.txt: Pythonの依存関係リスト。
.gitignore: Gitで追跡しないファイル・フォルダを指定。
README.md: プロジェクトの説明や使用方法を記載。
cdk.json: CDKの設定ファイル。
.env: 環境変数を定義するファイル（セキュリティに注意）。

まとめ

今回のガイドでは、AWS CDK（Python）を使用して構築したCRUD APIにAPIキー認証を導入する方法を詳しく解説しました。
IP制限に代わる認証手段として、APIキー認証はシンプルかつ効果的な方法であり、特定のユーザーやアプリケーションに対するアクセス制御を容易に実現できます。

ポイントまとめ:

APIキーの作成と管理: API GatewayでAPIキーを作成し、使用プランを設定することで、リクエストの制限やアクセス制御が可能になります。
API Gatewayの設定変更: 各APIメソッドに対してAPIキーの必須設定を追加することで、認証を強化します。
Lambda関数の柔軟な対応: 必要に応じて、Lambda関数内でAPIキーを取得・利用することが可能です。
セキュリティの向上: APIキーを適切に管理・配布することで、APIへの不正アクセスを防止します。

次のステップ:

APIキーのローテーション: 定期的にAPIキーを更新し、セキュリティを維持します。
使用プランのカスタマイズ: クライアントごとに異なる使用プランを設定し、リソースの適切な利用を促進します。
他の認証手法との併用: セキュリティ要件に応じて、APIキー認証と他の認証手法（例：JWT認証）を組み合わせることも検討してください。

このガイドを参考に、ぜひ自分のプロジェクトにAPIキー認証を導入し、より安全で制御されたAPIを提供してみてください。
必要に応じて、さらなる認証・認可の強化を行い、堅牢なAPIエコシステムを構築していきましょう！

参考資料

注記: セキュリティの観点から、.envファイルには機密情報を含めないようにし、必要に応じてAWS Secrets Managerなどのサービスを利用して機密情報を管理してください。また、APIキーは適切に管理し、不要になった場合は速やかに無効化・削除することを推奨します。

会社紹介

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

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