kmtnflower/backend/app/api/auth.py

"""Auth router: login, google OAuth, and current user info."""

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import joinedload
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import (
    create_access_token,
    get_current_user,
    verify_google_id_token,
    verify_password,
    hash_password,
)
from app.config import settings
from app.database import get_session
from app.user_models import User

router = APIRouter()


# ── Request/Response schemas ─────────────────────────────────

class LoginRequest(BaseModel):
    username: str
    password: str


class GoogleLoginRequest(BaseModel):
    id_token: str


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"


class UserInfoResponse(BaseModel):
    id: int
    email: str
    display_name: str
    avatar_url: str | None
    auth_provider: str
    is_admin: bool

    model_config = {"from_attributes": True}


# ── Routes ───────────────────────────────────────────────────

@router.post("/login", response_model=TokenResponse)
async def login(payload: LoginRequest):
    """Conditional admin login.

    - When NO admins exist: accepts hardcoded bootstrap credentials and
      auto-provisions the first admin user.
    - When admins DO exist: rejects hardcoded credentials (403) and requires
      normal password-based login for existing admin users.
    """
    if not settings.ADMIN_USERNAME or not settings.ADMIN_PASSWORD:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Bootstrap admin not configured")

    from app.database import async_session
    from app.user_models import Role, user_roles

    # Check if any admin users already exist in the DB
    async with async_session() as session:
        # Check for admin users: legacy is_admin flag OR admin role assignment
        admin_role_subq = (
            select(1)
            .select_from(user_roles)
            .join(Role, user_roles.c.role_id == Role.id)
            .where(user_roles.c.user_id == User.id)
            .where(Role.name == "admin")
            .exists()
        )

        result = await session.execute(
            select(1).where(
                (User.is_admin == True) | admin_role_subq
            ).limit(1)
        )
        has_admins = result.scalar() is not None

        if has_admins:
            # Admins exist — bootstrap creds are locked. Try normal password login.
            if (payload.username == settings.ADMIN_USERNAME and
                    payload.password == settings.ADMIN_PASSWORD):
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail="Bootstrap credentials disabled — admin users already exist",
                )

            # Normal password-based login for existing admin users
            result = await session.execute(
                select(User).options(joinedload(User.roles)).where(
                    (User.email == f"{payload.username}@local") |
                    (User.display_name == payload.username)
                )
            )
            user = result.unique().scalar_one_or_none()
            if user is None or not user.password_hash or not verify_password(payload.password, user.password_hash):
                raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
            if not user.is_admin_effective:
                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
            token = create_access_token(user.id, user.is_admin_effective)
            return TokenResponse(access_token=token)

        # Bootstrap path — no admins exist yet. Verify hardcoded creds.
        if payload.username != settings.ADMIN_USERNAME or payload.password != settings.ADMIN_PASSWORD:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

        # Ensure the 'admin' role exists
        result = await session.execute(select(Role).where(Role.name == "admin"))
        admin_role = result.scalar_one_or_none()
        if admin_role is None:
            admin_role = Role(name="admin", description="Full administrative access")
            session.add(admin_role)
            await session.commit()
            await session.refresh(admin_role)

        # Create the bootstrap admin user
        user = User(
            email=f"{settings.ADMIN_USERNAME}@local",
            display_name=settings.ADMIN_USERNAME,
            auth_provider="local",
            password_hash=hash_password(settings.ADMIN_PASSWORD),
            is_admin=True,
        )
        user.roles.append(admin_role)
        session.add(user)
        await session.commit()
        await session.refresh(user)

    token = create_access_token(user.id, user.is_admin_effective)
    return TokenResponse(access_token=token)


@router.post("/google", response_model=TokenResponse)
async def login_with_google(payload: GoogleLoginRequest, session: AsyncSession = Depends(get_session)):
    """Google OAuth sign-in. Accepts an ID token, verifies it, upserts user, returns JWT."""
    info = await verify_google_id_token(payload.id_token)

    result = await session.execute(select(User).where(User.email == info["email"]))
    user = result.scalar_one_or_none()

    if user is None:
        user = User(
            email=info["email"],
            display_name=info["name"],
            avatar_url=info["picture"] or None,
            auth_provider="google",
            is_admin=False,
        )
        session.add(user)
        await session.commit()
        await session.refresh(user)

    token = create_access_token(user.id, user.is_admin_effective)
    return TokenResponse(access_token=token)


@router.get("/me", response_model=UserInfoResponse)
async def me(current_user: User = Depends(get_current_user)):
    """Return current user info."""
    return current_user