今回のテーマ：Amazon Cognitoを用いたWebアプリケーションの認証機能実装

はじめまして！
TOPPANデジタルでAWS Top Engineersを目指し、日々成長中の村松です。
今回のテーマはAmazon Cognitoを用いた認証の実装です。
今回は、Webアプリケーションの認証機能をAmazon Cognitoを使って実装してみました。実装方法が皆さんの参考になればと思い、紹介いたします。

はじめに：Amazon Cognitoとは

Amazon Cognito（以下、Cognito）は、AWSが提供する認証、認可、ユーザー管理のためのサービスです。主にWebアプリケーションやモバイルアプリケーション（以下、アプリケーション）に、セキュアでスケーラブルな認証機能を簡単に追加することができます。これにより、ユーザーのサインインやサインアップ、パスワードリセットなどの機能を手軽に導入できます。

過去の記事では、Cognito以外でユーザー認証を実装している場合の多要素認証の設定方法を、Amazon Pinpointを使って紹介しました。 本記事では、Cognitoを利用したユーザー認証の方法についてご紹介します。

実現したかったこと

今回の目的は、以下の2つの要件を満たしつつアプリケーションにログイン認証機能を追加することです。

ⅰ. アプリケーション側の変更を最小限に抑えたユーザー認証の追加
ⅱ. ログインページとアプリケーション画面の切り替えによる認証実装

認証機能の実装概要

①Cognitoでユーザープールを作成し、アプリケーションクライアントを設定

Cognitoでユーザープールを作成し、アプリケーションクライアントを設定することで、ユーザーのログインや新規登録の方法をアプリケーションに合わせて設定できます。設定後、Cognitoから提供されるユーザープールIDとクライアントIDを使って、アプリケーションと接続できます。

②WGSIミドルウェアを用いてFlaskアプリケーションをFastAPIと統合

本件では、Flaskというウェブフレームワークを使ってアプリケーションを開発しています。アプリケーションに、FastAPIを統合するためにWSGIミドルウェアを使用しました。WSGIミドルウェアは、異なるウェブフレームワークを連携させるための仕組みです。

これにより、Cognitoによるユーザー認証を組み込んだログインページとアプリケーションの提供を、1つのアプリケーション内で切り替えて実行できるようになります。

# アプリケーションを提供するためのマウント
from fastapi import FastAPI
from starlette.middleware.wsgi import WSGIMiddleware
from your_flask_app import create_app  # Flaskアプリケーションをインポート

# FastAPIのインスタンスを作成
app = FastAPI()

# 静的ファイルを提供するためのマウント
app.mount("/static", StaticFiles(directory="static"), name="static")

# FlaskアプリケーションをWSGIミドルウェアを通してFastAPIに統合
flask_app = create_app(requests_pathname_prefix="/app/")
app.mount("/app", WSGIMiddleware(flask_app.server))

ログインページを表示
ルート（/）エンドポイントで、静的なログインページを表示するGETリクエストエンドポイントを作成しました。

@app.get("/", response_class=HTMLResponse)
async def read_root():
    with open("static/login.html") as f:
        return HTMLResponse(content=f.read())

ログイン用エンドポイント
ログインのためのPOSTリクエストエンドポイントを作成しました。
/loginエンドポイントでは、ユーザー名とパスワードを受け取って、authenticate_user関数でCognito認証を行います。認証が成功した場合、JWTトークンをクッキーに保存し、ユーザーはアプリケーションページにリダイレクトされます。認証が失敗した場合は、エラーメッセージが表示されます。

@app.post("/login")
async def login(id: str = Form(...), password: str = Form(...), request: Request = None):
    # ユーザーの認証を行う
    token = authenticate_user(id, password)
    if token:
        # トークンをクッキーに保存
        response = RedirectResponse(url="/app", status_code=HTTP_302_FOUND)
        response.set_cookie(key="access_token", value=token, secure=True, httponly=True)
        return response
    else:
        # 認証失敗の場合はエラーメッセージを返す
        return HTMLResponse(content="<p>Invalid credentials</p>", status_code=400)

③JWTトークンを用いた認証フローJWT

ログインページでユーザー認証(Cognito)を完了すると、アクセストークンが取得できます。さらにトークンの検証が成功すると、ユーザーはアプリケーションの画面にリダイレクトされ、アプリケーションが利用できます。

# 環境変数を設定
REGION = os.getenv("REGION")   # AWSリージョン
USER_POOL_ID = os.getenv("USER_POOL_ID")   # User Pool ID
CLIENT_ID = os.getenv("CLIENT_ID")   # App Client ID
PROFILE_NAME = os.getenv("PROFILE_NAME")   # AWS CLI用のプロファイル名


# JWTトークンを検証する関数
def verify_jwt(request: Request) -> dict:
    try:
        # クッキーからJWTトークンを取得
        token = request.cookies.get("access_token")
        if not token:
            print("Access token not found.")
            return None

        # JWTトークンのヘッダーをデコード
        header = token.split(".")[0]  # ヘッダー部分
        decoded_header = json.loads(base64.urlsafe_b64decode(header + "=="))  # デコード
        kid = decoded_header["kid"]  # kidを取得

        # JWKS（JSON Web Key Set）を取得するためのURLを作成
        jwks_url = f"https://cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}/.well-known/jwks.json"
        response = requests.get(jwks_url)
        jwks = response.json()

        # 一致する鍵を探す
        matching_key = None
        for key in jwks["keys"]:
            if key["kid"] == kid:
                matching_key = key
                break

        if not matching_key:
            print("No matching key found in JWKS.")
            return None

        # JWKから公開鍵を構築
        public_key = jwk.construct(matching_key)

        # JWTトークンをデコードして検証
        decoded_token = jwt.decode(token, public_key, algorithms=["RS256"], audience=CLIENT_ID)
        return decoded_token  # デコードされたトークンを返す

    except JWTError as e:
        print(f"JWT validation error: {str(e)}")
        return None
    except Exception as e:
        print(f"Unexpected error: Network error or missing configuration: {str(e)}")
        return None


# ユーザーをCognitoで認証する関数
def authenticate_user(username: str, password: str) -> str:
    # Boto3セッションを作成
    session = Session(profile_name=PROFILE_NAME)
    client = session.client("cognito-idp", region_name=REGION)

    try:
        # ユーザー名とパスワードを使って認証を試みる
        response = client.initiate_auth(
            ClientId=CLIENT_ID,
            AuthFlow="USER_PASSWORD_AUTH",
            AuthParameters={"USERNAME": username, "PASSWORD": password},
        )
        # 認証に成功した場合、JWTトークンを返す
        return response["AuthenticationResult"]["IdToken"]
    except ClientError as e:
        print("Authentication failed:", e.response["Error"]["Message"])
        return None


# JWT認証用のミドルウェア
class JWTMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path.startswith("app"):
            # リクエストオブジェクトを渡す
            if verify_jwt(request) is None:
                return RedirectResponse(url="/static/login.html", status_code=HTTP_302_FOUND)
        response = await call_next(request)
        return response


# JWTミドルウェアを追加して/appにアクセスする際に認証を確認する
app.add_middleware(JWTMiddleware)

所感：使ってみて

Cognitoを使うことで、アプリケーションのログイン認証機能を簡単に実装できました。とくに以下の2点が良かったです。

1. 手軽に認証機能の実装が可能
   Cognitoでユーザー登録、ログイン、パスワード管理を一元管理でき、認証機能を一から作る手間が省けました。

2. 柔軟な認証フロー
   ミドルウェアを活用し、認証ページの表示やログイン後のリダイレクトなど、アプリに最適な認証フローを簡単に実装できました。これにより、アプリケーション側の変更を最小限に抑えつつ、ログインページとアプリケーションの切り替えもスムーズに実現できました。

〆：さいごに

最後までお読みいただき、ありがとうございます！

今回、ユーザー認証機能をCognitoを用いて実装してみたところ、パスワード管理やトークン設定などが直感的で、スムーズに実装できました。

CognitoにはMFA認証やユーザー属性の活用など、まだまだ多くの機能がありますので、今後も検証を重ね、より良い実装の知見を増やしていきたいと思います！