moneyfyandClaude Opus 5
b586d27b77
- Anmeldung über Argon2id und JWT in httpOnly-Cookies, Refresh mit echter Rotation über die neue Tabelle refresh_token - AuthProvider-Protokoll als Vorbereitung für OIDC, Administrator-Anlage beim Erststart mit erzwungenem Passwortwechsel - CRUD für Konten, Kategorien (zweistufiger Baum), Firmen, Recurrences, Preisversionen, Buchungen, Budgets, Vorlagen und Sparziele - Fälligkeiten mit Overlay-Logik: abrufen, bestätigen, auslassen, zurücksetzen - Kontosalden zum Stichtag, Monatsübersicht mit Plan-Ist-Vergleich - SECRET_KEY jetzt mindestens 32 Zeichen; Platzhalter in Produktion abgelehnt - 61 neue Integrationstests, insgesamt 148 grün Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Wiederverwendbare FastAPI-Dependencies."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.cookies import ACCESS_COOKIE
|
|
from app.core.errors import AppError, AuthError
|
|
from app.core.security import decode_token
|
|
from app.db.session import get_session
|
|
from app.models import AppUser
|
|
from app.services.auth import get_user
|
|
|
|
DbSession = Annotated[AsyncSession, Depends(get_session)]
|
|
|
|
|
|
class PasswordChangeRequiredError(AppError):
|
|
"""Der Zugriff ist erst nach dem erzwungenen Passwortwechsel möglich."""
|
|
|
|
status_code = 403
|
|
code = "password_change_required"
|
|
|
|
|
|
async def get_current_user(request: Request, session: DbSession) -> AppUser:
|
|
"""Liest das Access-Token aus dem Cookie und lädt den Benutzer."""
|
|
token = request.cookies.get(ACCESS_COOKIE)
|
|
if not token:
|
|
raise AuthError("Nicht angemeldet.", code="not_authenticated")
|
|
|
|
payload = decode_token(token, "access")
|
|
try:
|
|
user_id = int(payload["sub"])
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise AuthError("Ungültiges Token.", code="invalid_token") from exc
|
|
|
|
user = await get_user(session, user_id)
|
|
if user is None:
|
|
raise AuthError("Der Benutzer existiert nicht mehr.", code="invalid_token")
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[AppUser, Depends(get_current_user)]
|
|
|
|
|
|
async def get_active_user(user: CurrentUser) -> AppUser:
|
|
"""Wie `get_current_user`, verlangt aber einen abgeschlossenen Passwortwechsel."""
|
|
if user.must_change_password:
|
|
raise PasswordChangeRequiredError(
|
|
"Das Passwort muss zuerst geändert werden.",
|
|
)
|
|
return user
|
|
|
|
|
|
ActiveUser = Annotated[AppUser, Depends(get_active_user)]
|