Johan LEROY
d7f775f9f7
Ajoute GET /auth/reset-password/validate (lecture seule, sans rate limit : le jeton est un secret de 256 bits non brute-forcable) pour que la page reset-password redirige immediatement vers /login si le lien est invalide ou expire, plutot que d'attendre la soumission du formulaire. La verification a la soumission (confirm_password_reset) reste la seule source de verite atomique.
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
# Piège : `consume()` est une seule instruction, sur le modèle de `claim_for_rotation()` du
|
|
# jeton de rafraîchissement. Un SELECT puis un UPDATE laisseraient une fenêtre où deux
|
|
# soumissions concurrentes du même lien réussiraient toutes les deux.
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.password_reset_token import PasswordResetToken
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ConsumedResetToken:
|
|
id: UUID
|
|
user_id: UUID
|
|
|
|
|
|
class PasswordResetTokenRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self._session = session
|
|
|
|
async def create(
|
|
self,
|
|
*,
|
|
user_id: UUID,
|
|
token_hash: bytes,
|
|
expires_at: datetime,
|
|
client_ip: str | None,
|
|
user_agent: str | None,
|
|
) -> PasswordResetToken:
|
|
jeton = PasswordResetToken(
|
|
user_id=user_id,
|
|
token_hash=token_hash,
|
|
expires_at=expires_at,
|
|
client_ip=client_ip,
|
|
user_agent=user_agent,
|
|
)
|
|
self._session.add(jeton)
|
|
await self._session.flush()
|
|
return jeton
|
|
|
|
async def consume(self, token_hash: bytes) -> ConsumedResetToken | None:
|
|
requete = (
|
|
update(PasswordResetToken)
|
|
.where(
|
|
PasswordResetToken.token_hash == token_hash,
|
|
PasswordResetToken.consumed_at.is_(None),
|
|
PasswordResetToken.expires_at > func.clock_timestamp(),
|
|
)
|
|
.values(consumed_at=func.clock_timestamp())
|
|
.returning(PasswordResetToken.id, PasswordResetToken.user_id)
|
|
)
|
|
ligne = (await self._session.execute(requete)).one_or_none()
|
|
if ligne is None:
|
|
return None
|
|
return ConsumedResetToken(id=ligne.id, user_id=ligne.user_id)
|
|
|
|
# Piège : simple SELECT, volontairement pas atomique avec la consommation. Sert seulement
|
|
# au feedback UX (jeton encore valide ?) ; `consume()` reste la seule source de vérité.
|
|
async def exists_valid(self, token_hash: bytes) -> bool:
|
|
requete = select(PasswordResetToken.id).where(
|
|
PasswordResetToken.token_hash == token_hash,
|
|
PasswordResetToken.consumed_at.is_(None),
|
|
PasswordResetToken.expires_at > func.clock_timestamp(),
|
|
)
|
|
return (await self._session.execute(requete)).first() is not None
|
|
|
|
async def invalidate_all_for_user(self, user_id: UUID) -> int:
|
|
resultat = await self._session.execute(
|
|
update(PasswordResetToken)
|
|
.where(PasswordResetToken.user_id == user_id, PasswordResetToken.consumed_at.is_(None))
|
|
.values(consumed_at=func.clock_timestamp())
|
|
.returning(PasswordResetToken.id)
|
|
)
|
|
return len(resultat.all())
|