90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
import time
|
|
|
|
from fastapi import APIRouter, Depends, Form, Request
|
|
|
|
from .. import totp
|
|
from ..users import MIN_PASSWORD_LEN, User, UserError
|
|
from .deps import FLASH_ERROR, FLASH_OK, SESSION_VERSION, current_user, flash, redirect, render
|
|
|
|
router = APIRouter()
|
|
|
|
TOTP_PENDING_TTL_SECONDS = 600
|
|
|
|
MSG_PASSWORD_MISMATCH = "Die neuen Passwörter stimmen nicht überein."
|
|
MSG_PASSWORD_WRONG = "Aktuelles Passwort falsch."
|
|
MSG_PASSWORD_CHANGED = "Passwort geändert."
|
|
MSG_TOTP_ENABLED = "Zwei-Faktor-Anmeldung aktiviert."
|
|
MSG_TOTP_DISABLED = "Zwei-Faktor-Anmeldung deaktiviert."
|
|
MSG_TOTP_INVALID = "Code ungültig."
|
|
MSG_TOTP_EXPIRED = "Einrichtung abgelaufen, bitte neu starten."
|
|
|
|
|
|
def _pending_secret(request: Request, user: User) -> str | None:
|
|
entry = request.app.state.totp_pending.get(user.id)
|
|
if entry is None:
|
|
return None
|
|
secret, expires = entry
|
|
if time.time() > expires:
|
|
request.app.state.totp_pending.pop(user.id, None)
|
|
return None
|
|
return secret
|
|
|
|
|
|
@router.get("/account")
|
|
def account(request: Request, user: User = Depends(current_user)):
|
|
secret = None if user.has_totp else _pending_secret(request, user)
|
|
uri = totp.provisioning_uri(secret, user.username) if secret else None
|
|
return render(request, "account.html", user=user, min_len=MIN_PASSWORD_LEN,
|
|
totp_secret=secret, totp_uri=uri)
|
|
|
|
|
|
@router.post("/account/password")
|
|
def change_password(request: Request, current: str = Form(""), new: str = Form(""),
|
|
confirm: str = Form(""), user: User = Depends(current_user)):
|
|
users = request.app.state.users
|
|
if new != confirm:
|
|
flash(request, MSG_PASSWORD_MISMATCH, FLASH_ERROR)
|
|
elif users.authenticate(user.username, current) is None:
|
|
flash(request, MSG_PASSWORD_WRONG, FLASH_ERROR)
|
|
else:
|
|
try:
|
|
users.set_password(user.id, new, actor=user.username)
|
|
except UserError as exc:
|
|
flash(request, str(exc), FLASH_ERROR)
|
|
else:
|
|
request.session[SESSION_VERSION] = users.get(user.id).session_version
|
|
flash(request, MSG_PASSWORD_CHANGED, FLASH_OK)
|
|
return redirect("/account")
|
|
|
|
|
|
@router.post("/account/totp/start")
|
|
def totp_start(request: Request, user: User = Depends(current_user)):
|
|
request.app.state.totp_pending[user.id] = (
|
|
totp.generate_secret(), time.time() + TOTP_PENDING_TTL_SECONDS)
|
|
return redirect("/account")
|
|
|
|
|
|
@router.post("/account/totp/confirm")
|
|
def totp_confirm(request: Request, code: str = Form(""), user: User = Depends(current_user)):
|
|
secret = _pending_secret(request, user)
|
|
if secret is None:
|
|
flash(request, MSG_TOTP_EXPIRED, FLASH_ERROR)
|
|
elif not totp.verify(secret, code):
|
|
flash(request, MSG_TOTP_INVALID, FLASH_ERROR)
|
|
else:
|
|
request.app.state.users.set_totp(user.id, secret, actor=user.username)
|
|
request.app.state.totp_pending.pop(user.id, None)
|
|
flash(request, MSG_TOTP_ENABLED, FLASH_OK)
|
|
return redirect("/account")
|
|
|
|
|
|
@router.post("/account/totp/disable")
|
|
def totp_disable(request: Request, password: str = Form(""), user: User = Depends(current_user)):
|
|
users = request.app.state.users
|
|
if users.authenticate(user.username, password) is None:
|
|
flash(request, MSG_PASSWORD_WRONG, FLASH_ERROR)
|
|
else:
|
|
users.set_totp(user.id, None, actor=user.username)
|
|
flash(request, MSG_TOTP_DISABLED, FLASH_OK)
|
|
return redirect("/account")
|