136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
from datetime import timedelta
|
|
|
|
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_optional_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.session_timeout_hours * 3600,
|
|
path="/",
|
|
)
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
async def dashboard(
|
|
request: Request,
|
|
user: User | None = Depends(get_optional_web_user),
|
|
):
|
|
# Anonymous visitors land on the public Scopes page instead of being
|
|
# bounced to login; the dashboard itself requires an account.
|
|
if user is None:
|
|
return RedirectResponse(url="/scopes", status_code=status.HTTP_302_FOUND)
|
|
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, timedelta(hours=settings.session_timeout_hours)
|
|
)
|
|
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, timedelta(hours=settings.session_timeout_hours)
|
|
)
|
|
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
|