58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.core.config import get_settings
|
|
from app.schemas.login import LoginRequest, LoginResponse
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest) -> LoginResponse:
|
|
settings = get_settings()
|
|
client_id = settings.authentik_client_id or settings.authentik_audience
|
|
|
|
if not settings.authentik_base_url or not client_id or not settings.authentik_client_secret:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="authentik_login_not_configured")
|
|
|
|
token_endpoint = settings.authentik_token_endpoint or urljoin(
|
|
settings.authentik_base_url.rstrip("/") + "/", "application/o/token/"
|
|
)
|
|
|
|
form = {
|
|
"grant_type": "password",
|
|
"client_id": client_id,
|
|
"client_secret": settings.authentik_client_secret,
|
|
"username": payload.username,
|
|
"password": payload.password,
|
|
"scope": "openid profile email",
|
|
}
|
|
|
|
try:
|
|
resp = httpx.post(
|
|
token_endpoint,
|
|
data=form,
|
|
timeout=10,
|
|
verify=settings.authentik_verify_tls,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="authentik_unreachable") from exc
|
|
|
|
if resp.status_code >= 400:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_username_or_password")
|
|
|
|
data = resp.json()
|
|
token = data.get("access_token")
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="authentik_missing_access_token")
|
|
|
|
return LoginResponse(
|
|
access_token=token,
|
|
token_type=data.get("token_type", "Bearer"),
|
|
expires_in=data.get("expires_in"),
|
|
scope=data.get("scope"),
|
|
)
|