from __future__ import annotations from typing import Any, Dict from authlib.integrations.starlette_client import OAuth from fastapi import APIRouter, HTTPException, Request from fastapi.responses import RedirectResponse from core.config import env_str OIDC_ISSUER = env_str("SPEECH_OIDC_ISSUER", "") OIDC_CLIENT_ID = env_str("SPEECH_OIDC_CLIENT_ID", "") OIDC_CLIENT_SECRET = env_str("SPEECH_OIDC_CLIENT_SECRET", "") OIDC_REDIRECT_URI = env_str("SPEECH_OIDC_REDIRECT_URI", "") SESSION_SECRET_KEY = env_str("SPEECH_SESSION_SECRET_KEY", "") oauth = OAuth() oauth.register( name="authentik", client_id=OIDC_CLIENT_ID, client_secret=OIDC_CLIENT_SECRET, server_metadata_url=f"{OIDC_ISSUER}.well-known/openid-configuration", client_kwargs={"scope": "openid email profile"}, ) router = APIRouter() @router.get("/login") async def login(request: Request, next: str = "/"): request.session["next"] = next if next.startswith("/") else "/" return await oauth.authentik.authorize_redirect(request, OIDC_REDIRECT_URI) @router.get("/callback") async def callback(request: Request): try: token = await oauth.authentik.authorize_access_token(request) except Exception as exc: raise HTTPException(status_code=401, detail=f"로그인 실패: {exc}") from exc userinfo = token.get("userinfo") or {} request.session["user"] = { "sub": userinfo.get("sub"), "username": userinfo.get("preferred_username") or userinfo.get("nickname"), "email": userinfo.get("email"), "name": userinfo.get("name"), } next_path = request.session.pop("next", "/") or "/" return RedirectResponse(url=next_path) @router.get("/logout") async def logout(request: Request): request.session.pop("user", None) return RedirectResponse(url="/") def require_login(request: Request) -> Dict[str, Any]: user = request.session.get("user") if not user: raise HTTPException( status_code=303, headers={"Location": f"/auth/login?next={request.url.path}"}, ) return user