665ff8a659
The project now covers both ASR and TTS, so "asr-v2" no longer fits. Renames: quadlet (asr-v2.container -> speech.container, container name, image tag), build.sh, FastAPI/UI title, and the OIDC env var prefix (ASR_V2_* -> SPEECH_*, matching the renamed Authentik provider/application slug "speech"). Internal Python packages (asr/, tts/, core/) were already named generically and don't change. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
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
|