"""Registration and login. See app/core/auth.py for hashing/token mechanics -- this layer owns the business rules (duplicate email, wrong credentials).""" from __future__ import annotations from sqlalchemy.ext.asyncio import AsyncSession from app.core.auth import ( burn_time_like_a_password_check, hash_password, normalize_email, verify_password, ) from app.models.user import User from app.repositories.user_repository import create_user, get_user_by_email class EmailAlreadyRegisteredError(Exception): pass class InvalidCredentialsError(Exception): pass async def register_user( session: AsyncSession, *, email: str, password: str, display_name: str | None = None ) -> User: normalized = normalize_email(email) if await get_user_by_email(session, normalized) is not None: raise EmailAlreadyRegisteredError(normalized) return await create_user( session, email=normalized, hashed_password=hash_password(password), display_name=display_name, ) async def authenticate_user(session: AsyncSession, *, email: str, password: str) -> User: normalized = normalize_email(email) user = await get_user_by_email(session, normalized) if user is None: # See burn_time_like_a_password_check's docstring: an email that # doesn't exist must fail no faster than a real one with a wrong # password, or response timing alone reveals which emails are # registered. burn_time_like_a_password_check() raise InvalidCredentialsError() if not user.is_active or not verify_password(password, user.hashed_password): raise InvalidCredentialsError() return user