ALB+Cognito認証フロー対応（バックエンド側改修）

app/api/auth.py

@@ -23,3 +23,18 @@ def get_current_user(request: Request):
         email=user.get("email"),
         role=user.get("role"),
     )
+
+
+from app.core.settings import get_settings
+from fastapi.responses import RedirectResponse
+
+@router.get("/login", summary="Login Redirect")
+def login_redirect():
+    """
+    ALB認証フローの起着点。
+    既にALBで認証されているため、フロントエンドのダッシュボードへリダイレクトする。
+    """
+    settings = get_settings()
+    # 末尾のスラッシュ調整などは必要に応じて行うが、基本は設定値を信頼
+    target_url = f"{settings.frontend_url}/dashboard"
+    return RedirectResponse(url=target_url, status_code=302)

app/core/middleware.py

@@ -27,18 +27,32 @@ async def dispatch(
             user_info["sub"] = oidc_identity
             if oidc_data:
                 try:
-                    # JWT payload is the second part
-                    payload_part = oidc_data.split(".")[1]
-                    # Add padding if needed
-                    payload_part += "=" * (-len(payload_part) % 4)
-                    decoded = base64.urlsafe_b64decode(payload_part)
-                    jwt_payload = json.loads(decoded)
-                    user_info["email"] = jwt_payload.get("email")
-                    user_info["username"] = jwt_payload.get(
-                        "username"
-                    ) or jwt_payload.get("cognito:username")
+                    # Format: header.payload.signature
+                    parts = oidc_data.split(".")
+                    if len(parts) > 1:
+                        payload_part = parts[1]
+                        # Add padding if needed
+                        payload_part += "=" * (-len(payload_part) % 4)
+                        decoded = base64.urlsafe_b64decode(payload_part)
+                        jwt_payload = json.loads(decoded)
+
+                        # Extract standard claims
+                        user_info["email"] = jwt_payload.get("email")
+                        # Cognito often provides 'cognito:username' or just 'username'
+                        user_info["username"] = (
+                            jwt_payload.get("username")
+                            or jwt_payload.get("cognito:username")
+                            or jwt_payload.get("email")  # Fallback to email as username
+                        )
+                        # Extract role if present (custom attribute)
+                        user_info["role"] = jwt_payload.get("custom:role") or "user"
                 except Exception as e:
                     print(f"Failed to decode OIDC data: {e}")
+                    # If decoding fails, we still have the identity (sub) from header
+                    if self.debug:
+                        import traceback
+                        traceback.print_exc()
+
         elif self.debug:
             # Local Development Mock
             user_info = {

app/core/settings.py

@@ -166,6 +166,9 @@ class AppSettings(BaseSettings):
     database_url: str = Field(
         default="sqlite:///./app_dev.sqlite3", alias="DATABASE_URL"
     )
+    frontend_url: str = Field(
+        default="http://localhost:3000", alias="FRONTEND_URL"
+    )
 
     aws: AWSSettings = Field(default_factory=AWSSettings)
     llm: LLMSettings = Field(default_factory=LLMSettings)