notify-bridge/packages/server/src/notify_bridge_server/api/users.py

"""User management API routes (admin only)."""

import logging

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from sqlalchemy import func
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

import bcrypt

from ..auth.dependencies import require_admin
from ..database.engine import get_session
from ..database.models import User

_LOGGER = logging.getLogger(__name__)

router = APIRouter(prefix="/api/users", tags=["users"])


class UserCreate(BaseModel):
    username: str
    password: str
    role: str = "user"


class UserUpdate(BaseModel):
    username: str | None = None
    password: str | None = None
    role: str | None = None


@router.get("")
async def list_users(
    admin: User = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
):
    """List all users (admin only)."""
    result = await session.exec(select(User))
    return [
        {"id": u.id, "username": u.username, "role": u.role, "created_at": u.created_at.isoformat()}
        for u in result.all()
    ]


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_user(
    body: UserCreate,
    admin: User = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
):
    """Create a new user (admin only)."""
    # Check for duplicate username
    result = await session.exec(select(User).where(User.username == body.username))
    if result.first():
        raise HTTPException(status_code=409, detail="Username already exists")

    if len(body.password) < 8:
        raise HTTPException(status_code=400, detail="Password must be at least 8 characters")

    user = User(
        username=body.username,
        hashed_password=bcrypt.hashpw(body.password.encode(), bcrypt.gensalt()).decode(),
        role=body.role if body.role in ("admin", "user") else "user",
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return {"id": user.id, "username": user.username, "role": user.role}


@router.patch("/{user_id}")
async def update_user(
    user_id: int,
    body: UserUpdate,
    admin: User = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
):
    """Update username and/or role for a user (admin only)."""
    user = await session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Track whether the identity that JWTs encode has changed. Any such change
    # must bump ``token_version`` so already-issued tokens are rejected — a
    # user demoted admin→user must not keep admin in their cached JWT until
    # expiry, and a rename should invalidate prior sessions too.
    identity_changed = False

    if body.username is not None and body.username != user.username:
        new_username = body.username.strip()
        if not new_username:
            raise HTTPException(status_code=400, detail="Username cannot be empty")
        dup = await session.exec(select(User).where(User.username == new_username))
        if dup.first():
            raise HTTPException(status_code=409, detail="Username already exists")
        user.username = new_username
        identity_changed = True

    if body.role is not None and body.role != user.role:
        if body.role not in ("admin", "user"):
            raise HTTPException(status_code=400, detail="Invalid role")
        # Prevent demoting the last admin. Done via a COUNT to avoid loading
        # every admin row; more importantly, re-checked *after* the role
        # change is staged (TOCTOU guard — two concurrent demotes can each
        # see admin_count=2 and both proceed, dropping to 0).
        if user.role == "admin" and body.role != "admin":
            admin_count = (await session.exec(
                select(func.count(User.id)).where(User.role == "admin")
            )).one()
            if isinstance(admin_count, tuple):
                admin_count = admin_count[0]
            if (admin_count or 0) <= 1:
                raise HTTPException(status_code=400, detail="Cannot demote the last admin")
        user.role = body.role
        identity_changed = True

    if identity_changed:
        user.token_version = (user.token_version or 1) + 1

    session.add(user)
    try:
        await session.commit()
    except Exception:
        await session.rollback()
        raise

    # Final defense against admin-count race: if we just demoted the last admin
    # due to a concurrent demote landing between our check and commit, undo.
    if body.role is not None and body.role != "admin":
        admin_count_after = (await session.exec(
            select(func.count(User.id)).where(User.role == "admin")
        )).one()
        if isinstance(admin_count_after, tuple):
            admin_count_after = admin_count_after[0]
        if (admin_count_after or 0) < 1:
            # Roll the user back to admin and re-commit.
            user.role = "admin"
            session.add(user)
            await session.commit()
            raise HTTPException(status_code=409, detail="Refused: would remove the last admin")

    await session.refresh(user)
    return {"id": user.id, "username": user.username, "role": user.role}


class ResetPasswordRequest(BaseModel):
    new_password: str


@router.put("/{user_id}/password")
async def reset_user_password(
    user_id: int,
    body: ResetPasswordRequest,
    admin: User = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
):
    """Reset a user's password (admin only)."""
    user = await session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    if len(body.new_password) < 8:
        raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
    user.hashed_password = bcrypt.hashpw(body.new_password.encode(), bcrypt.gensalt()).decode()
    # Invalidate all prior JWTs issued for this user — matches the self-serve
    # password-change path in auth/routes.py.
    user.token_version = (user.token_version or 1) + 1
    session.add(user)
    await session.commit()
    return {"success": True}


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: int,
    admin: User = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
):
    """Delete a user (admin only, cannot delete self)."""
    if user_id == admin.id:
        raise HTTPException(status_code=400, detail="Cannot delete yourself")
    user = await session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    await session.delete(user)
    await session.commit()