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
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""Passwort-Hashing und JWT-Ausstellung."""
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any, Literal
|
|
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHashError, VerifyMismatchError
|
|
|
|
from app.core.config import settings
|
|
from app.core.errors import AuthError
|
|
|
|
ALGORITHM = "HS256"
|
|
TokenType = Literal["access", "refresh"]
|
|
|
|
_hasher = PasswordHasher()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Erzeugt einen Argon2id-Hash."""
|
|
return _hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str | None) -> bool:
|
|
"""Prüft ein Passwort gegen den Hash. Fehlender Hash gilt immer als falsch."""
|
|
if not password_hash:
|
|
return False
|
|
try:
|
|
return _hasher.verify(password_hash, password)
|
|
except (VerifyMismatchError, InvalidHashError, ValueError):
|
|
return False
|
|
|
|
|
|
def needs_rehash(password_hash: str) -> bool:
|
|
"""True, wenn der Hash mit veralteten Parametern erzeugt wurde."""
|
|
try:
|
|
return _hasher.check_needs_rehash(password_hash)
|
|
except (InvalidHashError, ValueError):
|
|
return True
|
|
|
|
|
|
def create_token(
|
|
subject: int,
|
|
token_type: TokenType,
|
|
*,
|
|
expires_in: timedelta,
|
|
jti: str | None = None,
|
|
) -> tuple[str, str, datetime]:
|
|
"""Erzeugt ein signiertes JWT und liefert (Token, jti, Ablaufzeitpunkt)."""
|
|
now = datetime.now(UTC)
|
|
expires_at = now + expires_in
|
|
token_id = jti or uuid.uuid4().hex
|
|
payload: dict[str, Any] = {
|
|
"sub": str(subject),
|
|
"typ": token_type,
|
|
"jti": token_id,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expires_at.timestamp()),
|
|
"iss": settings.app_name,
|
|
}
|
|
token = jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)
|
|
return token, token_id, expires_at
|
|
|
|
|
|
def create_access_token(subject: int) -> tuple[str, str, datetime]:
|
|
return create_token(
|
|
subject, "access", expires_in=timedelta(minutes=settings.access_token_ttl_minutes)
|
|
)
|
|
|
|
|
|
def create_refresh_token(subject: int) -> tuple[str, str, datetime]:
|
|
return create_token(
|
|
subject, "refresh", expires_in=timedelta(days=settings.refresh_token_ttl_days)
|
|
)
|
|
|
|
|
|
def decode_token(token: str, expected_type: TokenType) -> dict[str, Any]:
|
|
"""Prüft Signatur, Ablauf und Tokenart. Wirft `AuthError` bei jedem Problem."""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.secret_key,
|
|
algorithms=[ALGORITHM],
|
|
issuer=settings.app_name,
|
|
options={"require": ["exp", "sub", "jti"]},
|
|
)
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise AuthError("Die Sitzung ist abgelaufen.", code="token_expired") from exc
|
|
except jwt.PyJWTError as exc:
|
|
raise AuthError("Ungültiges Token.", code="invalid_token") from exc
|
|
|
|
if payload.get("typ") != expected_type:
|
|
raise AuthError("Ungültige Tokenart.", code="invalid_token")
|
|
return payload
|