S.H.O.N.A.R._Desktop_Companion/backend/shonar/services/auth.py

"""Authentication service: registration, login, rotating refresh tokens with
reuse detection, logout, account deletion."""

from __future__ import annotations

import uuid
from datetime import UTC, datetime

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from shonar.core.security import (
    generate_refresh_token,
    hash_password,
    hash_refresh_token,
    refresh_token_ttl,
    verify_password,
)
from shonar.db.models import Device, RefreshToken, User, utcnow


class AuthError(Exception):
    """Safe, client-displayable auth failure (never leaks which half failed
    beyond what the flow requires)."""

    def __init__(self, message: str, status_code: int = 401):
        super().__init__(message)
        self.message = message
        self.status_code = status_code


async def register_user(
    session: AsyncSession, email: str, password: str, display_name: str | None
) -> User:
    email = email.strip().lower()
    existing = await session.scalar(select(User).where(User.email == email))
    if existing is not None:
        # Use a generic message; do not reveal whether the account exists in
        # flows where that matters. For self-hosted registration the UX cost
        # of "email already registered" is acceptable and helpful.
        raise AuthError("An account with this email already exists.", 409)
    user = User(
        email=email,
        password_hash=hash_password(password),
        display_name=display_name,
    )
    session.add(user)
    await session.flush()
    return user


async def issue_refresh_token(
    session: AsyncSession,
    user_id: uuid.UUID,
    family: uuid.UUID | None,
    device_id: uuid.UUID | None,
) -> tuple[str, RefreshToken]:
    token = generate_refresh_token()
    rt = RefreshToken(
        user_id=user_id,
        token_hash=hash_refresh_token(token),
        family=family or uuid.uuid4(),
        device_id=device_id,
        expires_at=datetime.now(UTC) + refresh_token_ttl(),
    )
    session.add(rt)
    await session.flush()
    return token, rt


async def login(
    session: AsyncSession,
    email: str,
    password: str,
    device_name: str | None,
    platform: str,
) -> tuple[User, str, Device]:
    """Returns (user, refresh_token, device). Raises AuthError safely."""
    email = email.strip().lower()
    user = await session.scalar(select(User).where(User.email == email))
    if user is None or user.deleted_at is not None or not user.is_active:
        raise AuthError("Invalid email or password.")
    if not verify_password(user.password_hash, password):
        raise AuthError("Invalid email or password.")

    device = Device(user_id=user.id, name=device_name or "Android device", platform=platform)
    session.add(device)
    await session.flush()

    refresh_token, _ = await issue_refresh_token(session, user.id, None, device.id)
    return user, refresh_token, device


async def rotate_refresh_token(
    session: AsyncSession, presented_token: str
) -> tuple[User, str, uuid.UUID | None]:
    """Consume a refresh token and issue a replacement in the same family.

    Reuse detection: presenting an already-consumed/revoked token revokes the
    entire family (an attacker's stolen token dies along with the real one).
    """
    token_hash = hash_refresh_token(presented_token)
    rt = await session.scalar(select(RefreshToken).where(RefreshToken.token_hash == token_hash))
    now = utcnow()

    if rt is None:
        raise AuthError("Invalid refresh token.")

    if rt.revoked_at is not None or rt.replaced_by is not None:
        # REUSE DETECTED — revoke the whole family. Commit BEFORE raising:
        # the request's transaction would otherwise roll back on the 401 and
        # silently undo the security-revocation.
        await session.execute(
            update(RefreshToken)
            .where(RefreshToken.family == rt.family, RefreshToken.revoked_at.is_(None))
            .values(revoked_at=now)
        )
        await session.commit()
        raise AuthError("Refresh token reuse detected. Please log in again.", 401)

    if rt.expires_at < now:
        raise AuthError("Refresh token expired.", 401)

    user = await session.get(User, rt.user_id)
    if user is None or user.deleted_at is not None or not user.is_active:
        raise AuthError("Account unavailable.", 401)

    new_token, new_rt = await issue_refresh_token(session, user.id, rt.family, rt.device_id)
    rt.revoked_at = now
    rt.replaced_by = new_rt.id

    if rt.device_id is not None:
        device = await session.get(Device, rt.device_id)
        if device is not None:
            device.last_seen_at = now
    await session.flush()
    return user, new_token, rt.device_id


async def logout(session: AsyncSession, presented_token: str) -> None:
    """Revoke the presented token's whole family (logs the device out)."""
    token_hash = hash_refresh_token(presented_token)
    rt = await session.scalar(select(RefreshToken).where(RefreshToken.token_hash == token_hash))
    if rt is None:
        return  # idempotent
    await session.execute(
        update(RefreshToken)
        .where(RefreshToken.family == rt.family, RefreshToken.revoked_at.is_(None))
        .values(revoked_at=utcnow())
    )


async def delete_account(session: AsyncSession, user: User, password: str) -> None:
    if not verify_password(user.password_hash, password):
        raise AuthError("Invalid password.", 403)
    now = utcnow()
    user.deleted_at = now
    user.is_active = False
    # Revoke every refresh token for the user.
    await session.execute(
        update(RefreshToken)
        .where(RefreshToken.user_id == user.id, RefreshToken.revoked_at.is_(None))
        .values(revoked_at=now)
    )
    # NOTE: hard deletion of rows/files is performed by the retention sweep
    # (services/retention.py) so an accidental deletion can be cancelled
    # within the grace window (see docs/security.md).