bike-app/apps/api/velodrome/auth/service.py

"""Auth bootstrap logic: register, login, session validation, logout.

Every function here runs against `db.auth_session()` (see its docstring for why these specific
pre-identity lookups need it) and every query is an exact match on a unique key — email,
token_hash, or code_hash — never an unfiltered scan.

Nothing outside this module should import `db.auth_session` — if a new feature needs it, that's
a sign the feature belongs here, not that the import should spread.
"""

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from uuid import UUID

from sqlalchemy import select, update

from velodrome.auth.security import (
    generate_token,
    hash_invite_code,
    hash_password,
    hash_token,
    verify_password,
)
from velodrome.config import get_settings
from velodrome.db import auth_session
from velodrome.models import ROLE_ADMIN, Invite, Session, User


class AuthError(Exception):
    """Base class for auth failures the API layer turns into 4xx responses."""


class InvalidCredentials(AuthError):
    pass


class InvalidInvite(AuthError):
    pass


class EmailAlreadyRegistered(AuthError):
    pass


class SessionInvalid(AuthError):
    pass


@dataclass(frozen=True)
class AuthenticatedSession:
    user_id: UUID
    email: str
    display_name: str
    # Carried here so the role a user actually has is available wherever identity is — an
    # authorization check on the first admin-only endpoint (realistically invite management) is
    # then a comparison against a value already in hand, not another query bolted on later. This
    # is data plumbing, not an authorization mechanism: nothing reads it yet, deliberately, since
    # there is no admin-only endpoint to protect (docs/DECISIONS.md D18). It is not a secret and
    # is not in any response model — `schemas.auth.UserOut` deliberately doesn't declare it, so
    # adding it here changes no HTTP response and no OpenAPI contract.
    role: str


async def register(
    *, email: str, password: str, display_name: str, invite_code: str
) -> AuthenticatedSession:
    """Validate an invite and create a user, atomically.

    Open signup does not exist — see docs/PLAN.md "Auth". What stops a shared invite link being
    redeemed twice concurrently is db.py's `BEGIN IMMEDIATE` setup, not a row lock on this
    SELECT — SQLite has no `SELECT ... FOR UPDATE` (SQLAlchemy's SQLite dialect silently no-ops
    `.with_for_update()`, confirmed empirically; it used to appear here when this ran against
    Postgres — see git history). `BEGIN IMMEDIATE` takes SQLite's write lock for the whole
    transaction up front, so two concurrent redemptions can't both read `used_count < max_uses`
    as true before either commits — the second one simply waits for the first transaction to
    finish, then sees the incremented count.
    """
    code_hash = hash_invite_code(invite_code)
    async with auth_session() as db:
        async with db.begin():
            invite = (
                await db.execute(select(Invite).where(Invite.code_hash == code_hash))
            ).scalar_one_or_none()

            if invite is None:
                raise InvalidInvite("invite code not found")
            if invite.revoked_at is not None:
                raise InvalidInvite("invite has been revoked")
            if invite.expires_at < datetime.now(UTC):
                raise InvalidInvite("invite has expired")
            if invite.used_count >= invite.max_uses:
                raise InvalidInvite("invite has already been used")
            if invite.email is not None and invite.email.lower() != email.lower():
                raise InvalidInvite("invite is pinned to a different email address")

            existing = (
                await db.execute(select(User).where(User.email == email))
            ).scalar_one_or_none()
            if existing is not None:
                raise EmailAlreadyRegistered("an account with this email already exists")

            user = User(
                email=email,
                display_name=display_name,
                password_hash=hash_password(password),
                role=invite.role,
            )
            db.add(user)
            await db.flush()  # populate user.id before we reference it below

            invite.used_count += 1

        return AuthenticatedSession(
            user_id=user.id,
            email=user.email,
            display_name=user.display_name,
            role=user.role,
        )


async def create_admin(*, email: str, password: str, display_name: str) -> AuthenticatedSession:
    """Create a user with the admin role, with no invite. Operator path only — see velodrome.cli.

    This is the one deliberate hole in "open signup does not exist": a fresh deployment has no
    users and therefore nobody who can issue the first invite, so the first account has to come
    from outside the HTTP API. It lives here rather than in cli.py because this is where
    `auth_session` belongs (see db.py's docstring — nothing outside this module imports it), and
    because the lookup below is exactly the pattern that module documents as safe: an exact match
    on a unique key, never a scan.

    It is not reachable over HTTP and never will be — nothing in `api/` calls it. Reaching it
    requires the ability to run a process inside the container, which is already the ability to
    read and rewrite the SQLite file directly, so it grants an operator-turned-attacker nothing
    they did not already have.

    Refuses outright if the email is taken, rather than updating the row — see docs/DECISIONS.md
    D18. Creating an account and resetting an existing account's password are different operations
    with different blast radii, and an operator re-running a bootstrap command they last ran
    months ago means the first, never the second. The check-then-insert is safe against a
    concurrent `register()` for the same email the same way invite redemption is: db.py issues
    `BEGIN IMMEDIATE`, so the two transactions serialize instead of interleaving, with the unique
    index on `users.email` as the backstop underneath that.
    """
    async with auth_session() as db:
        async with db.begin():
            existing = (
                await db.execute(select(User).where(User.email == email))
            ).scalar_one_or_none()
            if existing is not None:
                raise EmailAlreadyRegistered("an account with this email already exists")

            user = User(
                email=email,
                display_name=display_name,
                password_hash=hash_password(password),
                role=ROLE_ADMIN,
            )
            db.add(user)
            await db.flush()  # populate user.id before we reference it below

        return AuthenticatedSession(
            user_id=user.id,
            email=user.email,
            display_name=user.display_name,
            role=user.role,
        )


async def login(
    *, email: str, password: str, client: str, user_agent: str | None, ip: str | None
) -> tuple[str, AuthenticatedSession]:
    """Verify credentials and create a session. Returns (raw_token, session) — the raw token is
    handed to the caller exactly once; only its hash is ever stored.
    """
    async with auth_session() as db:
        async with db.begin():
            user = (await db.execute(select(User).where(User.email == email))).scalar_one_or_none()

            # Deliberately identical error for "no such user" and "wrong password" — this is the
            # one place a timing/response difference would leak which emails are registered.
            if user is None or not user.is_active:
                # Still run the hasher so this branch isn't measurably faster than a real
                # mismatch (argon2's own duration masks a database-lookup-only shortcut).
                verify_password(password, hash_password("decoy-password-never-matches"))
                raise InvalidCredentials("invalid email or password")
            if not verify_password(password, user.password_hash):
                raise InvalidCredentials("invalid email or password")

            settings = get_settings()
            raw_token, token_hash = generate_token()
            session_row = Session(
                user_id=user.id,
                token_hash=token_hash,
                client=client,
                user_agent=user_agent,
                ip=ip,
                expires_at=datetime.now(UTC) + timedelta(days=settings.session_ttl_days),
            )
            db.add(session_row)

        return raw_token, AuthenticatedSession(
            user_id=user.id,
            email=user.email,
            display_name=user.display_name,
            role=user.role,
        )


# Sessions are only touched this often to avoid a write on every single request; see
# docs/PLAN.md's auth section for the same throttling rule applied to last_seen_at.
_LAST_SEEN_THROTTLE = timedelta(minutes=5)


async def validate_session(raw_token: str) -> AuthenticatedSession:
    """Look a bearer token up and return the identity it belongs to, or raise SessionInvalid.

    This is the auth entrypoint for every authenticated request — it's what runs *before*
    db.scoped_session() can be used, because app.user_id isn't known until this returns.
    """
    token_hash = hash_token(raw_token)
    async with auth_session() as db:
        async with db.begin():
            row = (
                await db.execute(
                    select(Session, User)
                    .join(User, User.id == Session.user_id)
                    .where(Session.token_hash == token_hash)
                )
            ).one_or_none()

            if row is None:
                raise SessionInvalid("session not found")
            session_row, user = row

            if session_row.revoked_at is not None:
                raise SessionInvalid("session has been revoked")
            if session_row.expires_at < datetime.now(UTC):
                raise SessionInvalid("session has expired")
            if not user.is_active:
                raise SessionInvalid("account is disabled")

            now = datetime.now(UTC)
            if now - session_row.last_seen_at > _LAST_SEEN_THROTTLE:
                await db.execute(
                    update(Session).where(Session.id == session_row.id).values(last_seen_at=now)
                )

        return AuthenticatedSession(
            user_id=user.id,
            email=user.email,
            display_name=user.display_name,
            role=user.role,
        )


async def logout(raw_token: str) -> None:
    token_hash = hash_token(raw_token)
    async with auth_session() as db:
        async with db.begin():
            await db.execute(
                update(Session)
                .where(Session.token_hash == token_hash, Session.revoked_at.is_(None))
                .values(revoked_at=datetime.now(UTC))
            )