from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, status from fastapi.responses import HTMLResponse, RedirectResponse, Response from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.database import get_db from app.crud import user as user_crud from app.deps import get_current_web_user from app.models.user import User from app.services import ldap as ldap_service from app.services.auth import issue_token_pair from app.web.templates import templates router = APIRouter(include_in_schema=False) def _safe_next(next_url: str) -> str: """Return next_url only if it's a safe same-site path, else "/". Guards against open-redirect: rejects absolute URLs and protocol-relative ("//host", "/\\host") targets, and never bounces back to the auth pages. """ if ( next_url.startswith("/") and not next_url.startswith(("//", "/\\")) and not next_url.startswith(("/login", "/logout")) ): return next_url return "/" def _set_auth_cookie(response: Response, access_token: str) -> None: response.set_cookie( key=settings.cookie_name, value=access_token, httponly=True, secure=settings.cookie_secure, samesite="lax", max_age=settings.access_token_expire_minutes * 60, path="/", ) @router.get("/", response_class=HTMLResponse) async def dashboard( request: Request, user: User = Depends(get_current_web_user), ): return templates.TemplateResponse(request, "dashboard.html", {"user": user}) @router.get("/login", response_class=HTMLResponse) async def login_page(request: Request, next_url: str = Query("", alias="next")): return templates.TemplateResponse( request, "login.html", {"ldap_enabled": settings.ldap_enabled, "next_url": _safe_next(next_url)}, ) @router.post("/login") async def login_submit( request: Request, email: str = Form(...), password: str = Form(...), next_url: str = Form("", alias="next"), db: AsyncSession = Depends(get_db), ): user = await user_crud.authenticate(db, email, password) if user is None: return templates.TemplateResponse( request, "login.html", { "local_error": "Incorrect email or password", "email": email, "ldap_enabled": settings.ldap_enabled, "next_url": _safe_next(next_url), }, status_code=status.HTTP_401_UNAUTHORIZED, ) access, _ = await issue_token_pair(db, user) response = RedirectResponse( url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER ) _set_auth_cookie(response, access) return response @router.post("/login/ldap") async def login_ldap_submit( request: Request, username: str = Form(...), password: str = Form(...), next_url: str = Form("", alias="next"), db: AsyncSession = Depends(get_db), ): if not settings.ldap_enabled: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) user = await ldap_service.login(db, username, password) if user is None: return templates.TemplateResponse( request, "login.html", { "ldap_error": "Incorrect username or password", "username": username, "ldap_enabled": True, "next_url": _safe_next(next_url), }, status_code=status.HTTP_401_UNAUTHORIZED, ) access, _ = await issue_token_pair(db, user) response = RedirectResponse( url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER ) _set_auth_cookie(response, access) return response @router.post("/logout") async def logout(request: Request): response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER) response.delete_cookie(settings.cookie_name, path="/") return response