Google OpenID Connect - gosaaan1/hokulea-garage GitHub Wiki

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
from itsdangerous import URLSafeSerializer
import httpx

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="YOUR_SECRET_KEY")

# Google OAuth設定
GOOGLE_CLIENT_ID = "YOUR_GOOGLE_CLIENT_ID"
GOOGLE_CLIENT_SECRET = "YOUR_GOOGLE_CLIENT_SECRET"
GOOGLE_REDIRECT_URI = "http://localhost:8000/auth/callback"
GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/auth"
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"

# URLSafeSerializerの初期化
serializer = URLSafeSerializer("your-secret-key")

@app.get("/auth/login")
async def login(request: Request):
    # 認可リクエストのパラメータを設定
    params = {
        "client_id": GOOGLE_CLIENT_ID,
        "redirect_uri": GOOGLE_REDIRECT_URI,
        "scope": "openid email profile",
        "response_type": "code",
        "state": serializer.dumps({"redirect_url": request.query_params.get("redirect_url", "/")})
    }
    # Google認証URLにリダイレクト
    return RedirectResponse(f"{GOOGLE_AUTH_URL}?{'&'.join(f'{k}={v}' for k, v in params.items())}")

@app.get("/auth/callback")
async def auth_callback(code: str, state: str, request: Request):
    # stateパラメータをデコード
    state_data = serializer.loads(state)
    redirect_url = state_data.get("redirect_url", "/")

    # 認可コードをアクセストークンに交換
    token_params = {
        "client_id": GOOGLE_CLIENT_ID,
        "client_secret": GOOGLE_CLIENT_SECRET,
        "code": code,
        "grant_type": "authorization_code",
        "redirect_uri": GOOGLE_REDIRECT_URI
    }
    async with httpx.AsyncClient() as client:
        token_response = await client.post(GOOGLE_TOKEN_URL, data=token_params)
        token_data = token_response.json()

    if "error" in token_data:
        raise HTTPException(status_code=400, detail=token_data["error"])

    # アクセストークンを使用してユーザー情報を取得
    access_token = token_data["access_token"]
    async with httpx.AsyncClient() as client:
        userinfo_response = await client.get(
            GOOGLE_USERINFO_URL,
            headers={"Authorization": f"Bearer {access_token}"}
        )
        userinfo = userinfo_response.json()

    # セッションにユーザー情報を保存
    request.session["user"] = userinfo

    # クライアントにリダイレクト
    return RedirectResponse(url=redirect_url)

@app.get("/auth/user")
async def get_user(request: Request):
    user = request.session.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

@app.get("/auth/logout")
async def logout(request: Request):
    request.session.pop("user", None)
    return {"message": "Logged out successfully"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
from starlette_oauth2.client import OAuth2Client
from starlette_oauth2.middleware import OAuth2Middleware
from starlette.requests import Request

app = FastAPI()

# セッションミドルウェアの追加
app.add_middleware(SessionMiddleware, secret_key="YOUR_SECRET_KEY")

# Google OAuth2 クライアントの設定
google_oauth = OAuth2Client(
    client_id="YOUR_GOOGLE_CLIENT_ID",
    client_secret="YOUR_GOOGLE_CLIENT_SECRET",
    authorize_endpoint="https://accounts.google.com/o/oauth2/auth",
    token_endpoint="https://oauth2.googleapis.com/token",
    userinfo_endpoint="https://www.googleapis.com/oauth2/v2/userinfo",
    scope=["openid", "email", "profile"],
)

# OAuth2ミドルウェアの追加
app.add_middleware(
    OAuth2Middleware,
    clients={
        "google": google_oauth
    },
    redirect_uri="http://localhost:8000/auth/callback"
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/login")
async def login(request: Request):
    return await google_oauth.authorize(request, "http://localhost:8000/auth/callback")

@app.get("/auth/callback")
async def auth_callback(request: Request):
    token = await google_oauth.authorize_access_token(request)
    user = await google_oauth.parse_id_token(request, token)
    request.session["user"] = user
    return RedirectResponse(url="/")

@app.get("/user")
async def get_user(request: Request):
    user = request.session.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

@app.get("/logout")
async def logout(request: Request):
    request.session.pop("user", None)
    return {"message": "Logged out successfully"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)