100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
import time
|
|
|
|
from fastapi import APIRouter, Depends, Form, Request
|
|
|
|
from .. import audit, totp
|
|
from ..users import User
|
|
from .deps import (FLASH_ERROR, SESSION_UID, SESSION_VERSION, current_user, flash, redirect,
|
|
render)
|
|
|
|
router = APIRouter()
|
|
|
|
PENDING_UID = "pending_uid"
|
|
PENDING_TS = "pending_ts"
|
|
PENDING_TTL_SECONDS = 300
|
|
STATUS_UNAUTHORIZED = 401
|
|
STATUS_TOO_MANY = 429
|
|
|
|
MSG_LOGIN_FAILED = "Anmeldung fehlgeschlagen."
|
|
MSG_BLOCKED = "Zu viele Fehlversuche. Bitte später erneut versuchen."
|
|
MSG_TOTP_INVALID = "Code ungültig."
|
|
AUDIT_TARGET_MAX = 64
|
|
|
|
|
|
def _complete_login(request: Request, user: User):
|
|
request.session.clear()
|
|
request.session[SESSION_UID] = user.id
|
|
request.session[SESSION_VERSION] = user.session_version
|
|
audit.record(request.app.state.db, user.username, "login", user.username)
|
|
return redirect("/")
|
|
|
|
|
|
def _pending_uid(request: Request) -> int | None:
|
|
uid = request.session.get(PENDING_UID)
|
|
started = request.session.get(PENDING_TS, 0)
|
|
if not uid or time.time() - started > PENDING_TTL_SECONDS:
|
|
return None
|
|
return uid
|
|
|
|
|
|
@router.get("/login")
|
|
def login_form(request: Request):
|
|
return render(request, "login.html")
|
|
|
|
|
|
@router.post("/login")
|
|
def login(request: Request, username: str = Form(""), password: str = Form("")):
|
|
state = request.app.state
|
|
key = username.lower()
|
|
if state.limiter.blocked(key):
|
|
flash(request, MSG_BLOCKED, FLASH_ERROR)
|
|
return render(request, "login.html", status_code=STATUS_TOO_MANY)
|
|
user = state.users.authenticate(username, password)
|
|
if user is None:
|
|
state.limiter.record(key)
|
|
audit.record(state.db, username[:AUDIT_TARGET_MAX], "login.failed", username[:AUDIT_TARGET_MAX])
|
|
flash(request, MSG_LOGIN_FAILED, FLASH_ERROR)
|
|
return render(request, "login.html", status_code=STATUS_UNAUTHORIZED)
|
|
state.limiter.reset(key)
|
|
if user.has_totp:
|
|
request.session.clear()
|
|
request.session[PENDING_UID] = user.id
|
|
request.session[PENDING_TS] = time.time()
|
|
return redirect("/login/totp")
|
|
return _complete_login(request, user)
|
|
|
|
|
|
@router.get("/login/totp")
|
|
def totp_form(request: Request):
|
|
if _pending_uid(request) is None:
|
|
return redirect("/login")
|
|
return render(request, "login_totp.html")
|
|
|
|
|
|
@router.post("/login/totp")
|
|
def totp_submit(request: Request, code: str = Form("")):
|
|
uid = _pending_uid(request)
|
|
if uid is None:
|
|
return redirect("/login")
|
|
state = request.app.state
|
|
key = f"totp:{uid}"
|
|
if state.limiter.blocked(key):
|
|
request.session.clear()
|
|
flash(request, MSG_BLOCKED, FLASH_ERROR)
|
|
return render(request, "login.html", status_code=STATUS_TOO_MANY)
|
|
user = state.users.get(uid)
|
|
secret = state.users.totp_secret(uid)
|
|
if user is None or user.disabled or not secret or not totp.verify(secret, code):
|
|
state.limiter.record(key)
|
|
flash(request, MSG_TOTP_INVALID, FLASH_ERROR)
|
|
return render(request, "login_totp.html", status_code=STATUS_UNAUTHORIZED)
|
|
state.limiter.reset(key)
|
|
return _complete_login(request, user)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request, user: User = Depends(current_user)):
|
|
audit.record(request.app.state.db, user.username, "logout", user.username)
|
|
request.session.clear()
|
|
return redirect("/login")
|