"""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