API GatewayのAPIキーをSecrets Managerでうまく管理できないかを検討してみた

はじめに

API GatewayではAPIキーによる認証を設定できます。APIキーを必須としてデプロイしたAPIでは、配布したAPIキーをヘッダーでx-api-keyと指定してリクエストしないと403 Forbiddenで応答されます。

APIキーを使った認証は実装が簡単な反面、以下の課題があります：

• ユーザーごとに異なる認可スコープの実装
• アクティブなAPIキーの照合処理
• セキュリティ要件に準拠するためのキーローテーション管理

そこで、API GatewayのAPIキーをSecrets Managerに保存し、DynamoDBでユーザー権限を管理するアプローチを試してみました。これにより、Lambda Authorizerでの認証・認可処理や、プログラマティックなAPIキー運用が可能になります。

今回はキーローテーション管理の実装は、見送っています。
ユーザーからのAPIキー更新をトリガーに、API Gateway登録のAPIキー更新、Secrets Managerの更新処理は実現できると思います。

アーキテクチャ

本記事では、以下の構成でAPIキー管理システムを実装します：

• API GatewayでAPIキーを作成し、Secrets Managerに同じキーを保存
• DynamoDBでユーザーIDと権限情報を管理
• Lambda AuthorizerでAPIキーの検証とユーザー権限の確認を実施

実装内容

1. API GatewayにおけるAPIキー管理

以下、API GatewayでのAPIキー管理のスクリプト部分です。
この部分は以下のスクリプトを参考に、ツール化したものを想定しています。

APIキーの作成と使用量プランへの紐づけ

import boto3
import secrets

# APIキーの生成（セキュアなランダム文字列）
def generate_api_key(length=40):
    """セキュアなAPIキーを生成"""
    return secrets.token_urlsafe(length)[:length]

# API Gateway クライアントの作成
apigw = boto3.client('apigateway', region_name='ap-northeast-1')

# APIキーの作成
api_key_name = 'sample_api_key'
api_key_value = generate_api_key()

response = apigw.create_api_key(
    name=api_key_name,
    enabled=True,  # APIキーをアクティブにする
    value=api_key_value,
)

api_key_id = response['id']
print(f"APIキーID: {api_key_id}")
print(f"APIキー値: {api_key_value}")

# 使用量プランへの紐づけ
usage_plan_id = 'your-usage-plan-id'  # 既存の使用量プランID

response = apigw.create_usage_plan_key(
    usagePlanId=usage_plan_id,
    keyId=api_key_id,
    keyType="API_KEY",
)

print(f"使用量プランへの紐づけ完了: {response['id']}")

APIキーの一覧取得と検索

# 特定の名前のAPIキーを検索
def find_api_key_by_name(api_key_name):
    """APIキー名からAPIキーIDを取得"""
    response = apigw.get_api_keys(
        nameQuery=api_key_name,
        includeValues=False  # セキュリティのため値は取得しない
    )
    
    if response['items']:
        return response['items'][0]['id']
    return None

api_key_id = find_api_key_by_name('sample_api_key')
print(f"検索結果: {api_key_id}")

2. Secrets Managerでのキー管理

以下、Secrets Managerでのキー管理です。
こちらも以下スクリプトを参考にツール化したものを想定しています。

シークレットの作成

import boto3
import json
from botocore.exceptions import ClientError

# Secrets Manager クライアントの作成
secrets_client = boto3.client('secretsmanager', region_name='ap-northeast-1')

def create_api_key_secret(secret_name, api_key_value, api_key_id, scope=None):
    """
    Secrets ManagerにAPIキー情報を保存
    
    Args:
        secret_name: シークレット名
        api_key_value: APIキーの値
        api_key_id: API GatewayのAPIキーID
        scope: 認可スコープ（オプション）
    """
    secret_dict = {
        'api_key': api_key_value,
        'api_key_id': api_key_id,
        'scope': scope or ['read', 'write'],  # デフォルトスコープ
        'created_at': '2026-01-18T00:00:00Z'
    }
    
    try:
        response = secrets_client.create_secret(
            Name=secret_name,
            Description=f'API Gateway APIキー: {secret_name}',
            SecretString=json.dumps(secret_dict),
            Tags=[
                {'Key': 'Type', 'Value': 'APIKey'},
                {'Key': 'Service', 'Value': 'APIGateway'}
            ]
        )
        print(f"シークレット作成完了: {response['ARN']}")
        return response['ARN']
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceExistsException':
            print(f"シークレット '{secret_name}' は既に存在します")
        else:
            raise e

# 使用例
secret_arn = create_api_key_secret(
    secret_name='apigateway_apikey_user1',
    api_key_value=api_key_value,
    api_key_id=api_key_id,
    scope=['read', 'write', 'delete']
)

シークレットの取得

def get_api_key_from_secrets(secret_name):
    """
    Secrets ManagerからAPIキー情報を取得
    
    Args:
        secret_name: シークレット名
        
    Returns:
        dict: APIキー情報
    """
    try:
        response = secrets_client.get_secret_value(SecretId=secret_name)
        secret_string = response['SecretString']
        secret_dict = json.loads(secret_string)
        return secret_dict
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'ResourceNotFoundException':
            print(f"シークレット '{secret_name}' が見つかりません")
        elif error_code == 'InvalidRequestException':
            print(f"リクエストが無効です: {e}")
        elif error_code == 'InvalidParameterException':
            print(f"パラメータが無効です: {e}")
        else:
            raise e
        return None

# 使用例
secret_data = get_api_key_from_secrets('apigateway_apikey_user1')
if secret_data:
    print(f"APIキー: {secret_data['api_key'][:10]}...")
    print(f"スコープ: {secret_data['scope']}")

シークレットの更新

import uuid

def update_api_key_secret(secret_name, new_api_key_value, new_api_key_id):
    """
    Secrets ManagerのAPIキー情報を更新
    
    Args:
        secret_name: シークレット名
        new_api_key_value: 新しいAPIキーの値
        new_api_key_id: 新しいAPI GatewayのAPIキーID
    """
    # 既存のシークレット情報を取得
    existing_secret = get_api_key_from_secrets(secret_name)
    
    if not existing_secret:
        print("既存のシークレットが見つかりません")
        return None
    
    # 新しい情報で更新
    updated_secret = {
        'api_key': new_api_key_value,
        'api_key_id': new_api_key_id,
        'scope': existing_secret.get('scope', ['read', 'write']),
        'updated_at': '2026-01-18T12:00:00Z'
    }
    
    try:
        response = secrets_client.update_secret(
            SecretId=secret_name,
            ClientRequestToken=str(uuid.uuid4()),  # 冪等性のためのトークン
            SecretString=json.dumps(updated_secret)
        )
        print(f"シークレット更新完了: {response['ARN']}")
        return response
    except ClientError as e:
        print(f"更新エラー: {e}")
        raise e

# 使用例
new_key = generate_api_key()
update_api_key_secret(
    secret_name='apigateway_apikey_user1',
    new_api_key_value=new_key,
    new_api_key_id='new-api-key-id'
)

3. DynamoDBでのユーザー権限（認可）管理

ユーザーIDごとの権限情報をDynamoDBで管理します。

テーブル設計

• テーブル名: user_permissions
• パーティションキー: user_id (String)
• プロパティ例

{
  "user_id": "user001",
  "permissions": ["read", "write"],
  "secret_name": "apigateway_apikey_user001",
  "api_key_id": "abc123xyz",
  "created_at": "2026-01-18T00:00:00Z",
  "updated_at": "2026-01-18T00:00:00Z"
}

ユーザー権限の操作

以下DyanmoDBへのユーザー権限の設定例です。

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
table = dynamodb.Table('user_permissions')

def register_user_permission(user_id, secret_name, api_key_id, permissions):
    """DynamoDBにユーザー権限を登録"""
    timestamp = datetime.utcnow().isoformat() + 'Z'
    
    table.put_item(
        Item={
            'user_id': user_id,
            'secret_name': secret_name,
            'api_key_id': api_key_id,
            'permissions': permissions,
            'created_at': timestamp,
            'updated_at': timestamp
        }
    )
    print(f"ユーザー {user_id} の権限を登録しました")

def get_user_permissions(user_id):
    """DynamoDBからユーザー権限を取得"""
    response = table.get_item(Key={'user_id': user_id})
    return response.get('Item')

# 使用例
register_user_permission(
    user_id='user001',
    secret_name='apigateway_apikey_user001',
    api_key_id='abc123xyz',
    permissions=['read', 'write']
)

4. Lambda Authorizerでの認証・認可処理

Lambda Authorizerを使用して、APIキーの検証とユーザー権限の確認を行います。

import json
import boto3
from botocore.exceptions import ClientError

secrets_client = boto3.client('secretsmanager')
dynamodb = boto3.resource('dynamodb')
user_permissions_table = dynamodb.Table('user_permissions')

def lambda_handler(event, context):
    """
    Lambda Authorizer: APIキーの検証とユーザー権限の確認
    
    処理フロー:
    1. リクエストヘッダーからAPIキーとユーザーIDを取得
    2. DynamoDBからユーザー情報を取得
    3. Secrets ManagerでAPIキーを検証
    4. 権限情報を含むIAMポリシーを返却
    """
    # リクエストヘッダーから情報を取得
    headers = event.get('headers', {})
    api_key = headers.get('x-api-key')
    user_id = headers.get('x-user-id')
    method_arn = event['methodArn']
    
    if not api_key or not user_id:
        print("APIキーまたはユーザーIDが提供されていません")
        raise Exception('Unauthorized')
    
    try:
        # DynamoDBからユーザー情報を取得
        user_response = user_permissions_table.get_item(Key={'user_id': user_id})
        
        if 'Item' not in user_response:
            print(f"ユーザー {user_id} が見つかりません")
            raise Exception('Unauthorized')
        
        user_data = user_response['Item']
        secret_name = user_data['secret_name']
        expected_permissions = user_data['permissions']
        
        # Secrets ManagerからAPIキー情報を取得
        secret_response = secrets_client.get_secret_value(SecretId=secret_name)
        secret_data = json.loads(secret_response['SecretString'])
        
        # APIキーの検証
        if secret_data['api_key'] != api_key:
            print(f"ユーザー {user_id} のAPIキーが一致しません")
            raise Exception('Unauthorized')
        
        # 権限情報を取得（Secrets Managerとユーザー権限の積集合）
        secret_scope = secret_data.get('scope', [])
        allowed_permissions = list(set(secret_scope) & set(expected_permissions))
        
        print(f"ユーザー {user_id} の認証成功")
        print(f"許可された権限: {allowed_permissions}")
        
        # IAMポリシーを生成
        return generate_policy(
            principal_id=user_id,
            effect='Allow',
            resource=method_arn,
            context={
                'user_id': user_id,
                'permissions': json.dumps(allowed_permissions),
                'api_key_id': secret_data['api_key_id']
            }
        )
        
    except ClientError as e:
        print(f"AWS サービスエラー: {e}")
        raise Exception('Unauthorized')
    except Exception as e:
        print(f"認証エラー: {e}")
        raise Exception('Unauthorized')

def generate_policy(principal_id, effect, resource, context=None):
    """IAMポリシードキュメントを生成"""
    policy = {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': effect,
                'Resource': resource
            }]
        }
    }
    
    if context:
        policy['context'] = context
    
    return policy

まとめ

API Gateway、Secrets Manager、DynamoDBを組み合わせることで、セキュアで柔軟なAPIキー管理システムを構築できました。

実現できたこと

• 3サービス連携による一元管理: API Gateway、Secrets Manager、DynamoDBでAPIキーとユーザー権限を統合管理
• ユーザーベースの認可: DynamoDBでユーザーごとの権限を管理し、Lambda Authorizerで検証
• きめ細かいアクセス制御: Secrets Managerとユーザー権限の両方を考慮した認可スコープの実装
• 監査証跡の確保: Secrets Managerのバージョン管理とDynamoDBでキーの変更履歴を追跡

注意点

• パフォーマンス: Lambda AuthorizerでSecrets ManagerとDynamoDBを呼び出すため、レイテンシが増加します。キャッシュの活用を検討してください
• コスト: Secrets Managerはシークレット数とAPI呼び出し回数、DynamoDBは読み取り/書き込みキャパシティに応じて課金されます
• 整合性の維持: 3つのサービス間でデータの整合性を保つため、適切なエラーハンドリングが必要です
• セキュリティ: Lambda関数には適切なIAMロールを設定し、最小権限の原則に従ってください

この構成により、認証・認可をカバーする実用的なAPIキー管理ソリューションを実現できます。