コンテンツにスキップ

認証システム

LambAPI では、DynamoDB を使用した JWT ベースの認証システムを提供します。この認証システムは、ユーザー管理、セッション管理、ロールベースアクセス制御をサポートしています。

概要

特徴

  • JWT トークン認証: セキュアなトークンベース認証
  • DynamoDB バックエンド: AWS サービスとの完全統合
  • カスタマイズ可能: BaseUser を継承してカスタムユーザーモデルを作成
  • ロールベース認証: 細かいアクセス制御をサポート
  • セッション管理: DynamoDB を使用した永続セッション
  • パスワード暗号化: bcrypt による安全なハッシュ化

アーキテクチャ

graph TD
    A[Client] --> B[API Endpoint]
    B --> C[DynamoDBAuth]
    C --> D[BaseUser/CustomUser]
    C --> E[DynamoDB]
    C --> F[JWT Token]
    F --> G[Session Storage]

インストール

認証機能を使用するには、オプショナル依存関係をインストールします:

pip install lambapi[auth]

必要な依存関係: - boto3>=1.28.0 - DynamoDB 接続 - PyJWT>=2.8.0 - JWT トークン処理 - bcrypt>=4.0.0 - パスワードハッシュ化 - cryptography>=41.0.0 - 暗号化サポート

基本的な使用方法

1. BaseUser の使用

最もシンプルな使用方法:

from lambapi import API, create_lambda_handler
from lambapi.auth import BaseUser, DynamoDBAuth

def create_app(event, context):
    app = API(event, context)

    # 認証システムの初期化(secret_key が必須)
    auth = DynamoDBAuth(secret_key="your-secure-secret-key")
    # または環境変数を設定: export LAMBAPI_SECRET_KEY="your-secure-secret-key"
    # auth = DynamoDBAuth()  # 環境変数 LAMBAPI_SECRET_KEY から自動取得

    @app.post("/auth/signup")
    def signup(request):
        return auth.signup(request)

    @app.post("/auth/login")
    def login(request):
        return auth.login(request)

    @app.post("/auth/logout")
    def logout(request):
        return auth.logout(request)

    @app.get("/protected")
    def protected_endpoint(request):
        user = auth.get_authenticated_user(request)
        return {"message": f"Hello, {user.id}!"}

    return app

lambda_handler = create_lambda_handler(create_app)

2. カスタムユーザーモデル

より複雑なユーザー情報が必要な場合:

from lambapi.auth import BaseUser, DynamoDBAuth

class User(BaseUser):
    class Meta(BaseUser.Meta):
        table_name = "my_users"
        secret_key = "your-secret-key-here"  # 本番環境では環境変数を使用
        is_email_login = True
        is_role_permission = True

    def __init__(self, id, password, name="", email="", role="user"):
        super().__init__(id, password)
        self.name = name
        self.email = email
        self.role = role

# カスタムユーザーで認証システムを初期化(secret_key 必須)
auth = DynamoDBAuth(User, secret_key="your-secure-secret-key")

3. ルーターを使用した認証エンドポイント

認証エンドポイントを自動的に作成:

from lambapi import API
from lambapi.auth import BaseUser, DynamoDBAuth, create_auth_router

def create_app(event, context):
    app = API(event, context)

    # 認証システムと関連ルーターを作成
    auth = DynamoDBAuth(User, secret_key="your-secure-secret-key")
    auth_router = create_auth_router(auth)

    # 認証ルーターを登録
    app.include_router(auth_router)

    return app

これにより以下のエンドポイントが自動作成されます: - POST /auth/signup - ユーザー登録 - POST /auth/login - ログイン - POST /auth/logout - ログアウト - DELETE /auth/user/{user_id} - ユーザー削除 - PUT /auth/user/{user_id}/password - パスワード更新

ロールベースアクセス制御

require_role デコレータ

特定のロールを持つユーザーのみにアクセスを制限:

@app.get("/admin/users")
@auth.require_role("admin")
def admin_only(user, request):
    # user パラメータが自動注入される
    return {"message": f"Admin access granted to {user.id}"}

@app.get("/moderator/reports")
@auth.require_role(["admin", "moderator"])
def moderator_access(user, request):
    # 複数のロールを許可
    return {"reports": [...]}

Authenticated 依存性注入

lambapi v0.2.1 以降では、依存性注入を使用して、より型安全で簡潔な認証処理が可能です:

from lambapi import API, Authenticated, Query, Path

@app.get("/profile")
@auth.require_role("user")
def get_profile(
    user: CustomUser = Authenticated(..., description="認証されたユーザー")
):
    # user パラメータが自動的に注入される
    return {
        "user_id": user.id,
        "role": getattr(user, 'role', 'user'),
        "created_at": user.created_at.isoformat() if user.created_at else None
    }

@app.post("/admin/users/{target_user_id}")
@auth.require_role("admin")
def update_user_as_admin(
    # 複数の依存性注入を組み合わせ可能
    admin: CustomUser = Authenticated(..., description="管理者ユーザー"),
    target_user_id: str = Path(..., description="対象ユーザー ID"),
    new_role: str = Query(..., description="新しいロール")
):
    return {
        "message": f"管理者 {admin.id} がユーザー {target_user_id} のロールを {new_role} に変更しました"
    }

従来方式との比較

# 従来の方式(引き続きサポート)
@app.get("/profile")
@auth.require_role("user")  
def get_profile_legacy(user, request):
    return {"user_id": user.id}

# 新しい依存性注入方式
@app.get("/profile")
@auth.require_role("user")
def get_profile_modern(
    user: CustomUser = Authenticated(...)
):
    return {"user_id": user.id}

新しい方式の利点: - 型安全性: ユーザーオブジェクトの型が明確 - IDE サポート: 自動補完や型チェック - バリデーション: パラメータの自動バリデーション - ドキュメント生成: 自動 API 仕様書生成

手動認証チェック

より柔軟な認証制御:

@app.get("/profile")
def get_profile(request):
    try:
        user = auth.get_authenticated_user(request)
        return {"profile": user.to_dict()}
    except AuthenticationError:
        return {"error": "Authentication required"}, 401

設定オプション

Meta クラス設定

BaseUser の Meta クラスで動作をカスタマイズ:

class User(BaseUser):
    class Meta(BaseUser.Meta):
        # DynamoDB 設定
        table_name = "users"                    # テーブル名
        endpoint_url = "http://localhost:8000"  # ローカル DynamoDB

        # JWT 設定
        secret_key = "your-secret-key"          # JWT 署名キー
        expiration = 3600                       # トークン有効期限(秒)

        # 機能設定
        is_email_login = True                   # メールログインを有効化
        is_role_permission = True               # ロール権限を有効化
        enable_auth_logging = False             # 認証ログを有効化

        # ID 設定
        id_type = "uuid"                        # UUID 自動生成

        # パスワード要件
        password_min_length = 8                 # 最小文字数
        password_require_uppercase = False      # 大文字必須
        password_require_lowercase = False      # 小文字必須
        password_require_digit = True           # 数字必須
        password_require_special = False        # 特殊文字必須

        # タイムスタンプ
        auto_timestamps = True                  # 自動タイムスタンプ

API リファレンス

ユーザー登録

POST /auth/signup
Content-Type: application/json

{
  "id": "user123",
  "password": "password123",
  "email": "user@example.com",     // is_email_login=True の場合必須
  "name": "User Name",             // カスタムフィールド
  "role": "user"                   // is_role_permission=True の場合
}

レスポンス例:

{
  "message": "ユーザー登録が完了しました",
  "user_id": "user123"
}

ログイン

POST /auth/login
Content-Type: application/json

{
  "id": "user123",                 // ID またはメールでログイン
  "password": "password123"
}

レスポンス例:

{
  "message": "ログインしました",
  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "user": {
    "id": "user123",
    "email": "user@example.com",
    "name": "User Name",
    "role": "user"
  }
}

ログアウト

POST /auth/logout
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

レスポンス例:

{
  "message": "ログアウトしました"
}

認証が必要なエンドポイント

GET /protected-endpoint
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

DynamoDB テーブル設計

テーブル構造

{
  "TableName": "users",
  "KeySchema": [
    {
      "AttributeName": "id",
      "KeyType": "HASH"
    }
  ],
  "AttributeDefinitions": [
    {
      "AttributeName": "id",
      "AttributeType": "S"
    },
    {
      "AttributeName": "email",
      "AttributeType": "S"
    }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "email-index",
      "KeySchema": [
        {
          "AttributeName": "email",
          "KeyType": "HASH"
        }
      ],
      "Projection": {
        "ProjectionType": "ALL"
      }
    }
  ]
}

セッション管理

セッション情報は同じテーブルに TTL 付きで保存されます:

{
  "id": "abc123def456",      // セッション ID(16 文字のハッシュ)
  "token": "eyJhbGc...",     // JWT トークン
  "user_id": "user123",      // ユーザー ID
  "exp": "2024-01-01T12:00:00Z",  // 有効期限
  "ttl": 1704110400          // DynamoDB TTL(自動削除)
}

セキュリティ考慮事項

パスワード保護

  • bcrypt ハッシュ化: ソルト付きハッシュで保存
  • 設定可能な要件: 文字数、文字種別の制限
  • フォールバック: bcrypt が利用できない場合は SHA-256(テスト用)

トークンセキュリティ

  • JWT 署名: HMAC-SHA256 で署名
  • 有効期限: 設定可能なトークン有効期限
  • セッション検証: トークンと DynamoDB セッションの二重チェック

推奨事項

  1. 環境変数を使用 (最重要):

    # 本番環境での推奨方法
    export LAMBAPI_SECRET_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
    
    # コード内では環境変数から自動取得
    auth = DynamoDBAuth()  # LAMBAPI_SECRET_KEY を自動使用
    

  2. 明示的な secret_key 指定 (開発・テスト用):

    # 開発環境やテストでの明示的指定
    auth = DynamoDBAuth(secret_key="development-key-do-not-use-in-production")
    

  3. 強力な秘密鍵を生成:

    # 安全なランダムキー生成
    python -c "import secrets; print(secrets.token_urlsafe(32))"
    

  4. 秘密鍵の管理:

  5. 絶対にソースコードに含めない
  6. 環境変数または AWS Systems Manager Parameter Store を使用
  7. 定期的にローテーションを実施

  8. HTTPS の使用: 本番環境では必ず HTTPS を使用

  9. 適切な権限設定: DynamoDB IAM ロールの最小権限の原則

トラブルシューティング

よくあるエラー

ImportError: 認証依存関係がない

# 解決方法
pip install lambapi[auth]

DynamoDB 接続エラー

# ローカル開発用設定
class User(BaseUser):
    class Meta(BaseUser.Meta):
        endpoint_url = "http://localhost:8000"

JWT デコードエラー

# secret_key の一致を確認
# トークンの有効期限を確認

ログの有効化

class User(BaseUser):
    class Meta(BaseUser.Meta):
        enable_auth_logging = True

# ログ出力例
# Auth Event: {"event": "login_success", "timestamp": "2024-01-01T12:00:00", "user_id": "user123"}

サンプルコード

完全な認証付き API

import os
from lambapi import API, create_lambda_handler
from lambapi.auth import BaseUser, DynamoDBAuth, create_auth_router
from lambapi.exceptions import AuthenticationError

class User(BaseUser):
    class Meta(BaseUser.Meta):
        table_name = os.getenv("DYNAMODB_TABLE", "users")
        secret_key = os.getenv("JWT_SECRET", "dev-secret")
        expiration = 3600  # 1 時間
        is_email_login = True
        is_role_permission = True
        enable_auth_logging = True

    def __init__(self, id, password, name="", email="", role="user"):
        super().__init__(id, password)
        self.name = name
        self.email = email
        self.role = role

def create_app(event, context):
    app = API(event, context)

    # 認証システムの初期化
    auth = DynamoDBAuth(User)

    # 認証エンドポイントの追加
    auth_router = create_auth_router(auth)
    app.include_router(auth_router)

    # パブリックエンドポイント
    @app.get("/")
    def public_endpoint():
        return {"message": "Public access"}

    # 認証が必要なエンドポイント
    @app.get("/profile")
    def get_profile(request):
        user = auth.get_authenticated_user(request)
        return {"profile": user.to_dict()}

    # ロール制限エンドポイント
    @app.get("/admin/stats")
    @auth.require_role("admin")
    def admin_stats(user, request):
        return {"stats": "admin only data", "user": user.id}

    # カスタム認証チェック
    @app.put("/profile")
    def update_profile(request):
        try:
            user = auth.get_authenticated_user(request)
            data = request.json()

            # プロフィール更新ロジック
            user.update_attributes(name=data.get("name"))

            return {"message": "Profile updated"}
        except AuthenticationError:
            return {"error": "Authentication required"}, 401

    return app

lambda_handler = create_lambda_handler(create_app)

テスト用コード

import unittest
import json
from lambapi import Request
from lambapi.auth import BaseUser, DynamoDBAuth

class TestAuth(unittest.TestCase):
    def setUp(self):
        self.auth = DynamoDBAuth(BaseUser)

    def test_user_signup(self):
        # ユーザー登録テスト
        event = {
            "body": json.dumps({
                "id": "testuser",
                "password": "password123"
            })
        }
        request = Request(event)
        result = self.auth.signup(request)

        self.assertEqual(result["user_id"], "testuser")
        self.assertIn("message", result)

参考資料