mule-image/backend/app/routers/auth.py

"""
Authentication router — login, token refresh, profile, first-run setup,
and optional OIDC (Authentik) sign-in.
"""
import os
import re
import secrets
import logging
from typing import Optional
from urllib.parse import urlencode, urlparse

from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import RedirectResponse
from pydantic import BaseModel
from sqlalchemy import select, func as sa_func
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import hash_password, verify_password, create_access_token, create_refresh_token, decode_token
from app.auth_oidc import get_client as get_oidc_client, is_enabled as oidc_is_enabled, provider_label, PROVIDER_NAME
from app.database import get_db
from app.dependencies import get_current_user
from app.models.user import User
from app.models.folders import SourceRoot
from app.services.gravatar import gravatar_url
from app.config import settings

logger = logging.getLogger(__name__)

router = APIRouter()


# ---------------------------------------------------------------------------
# Request / response schemas
# ---------------------------------------------------------------------------

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

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

class RefreshRequest(BaseModel):
    refresh_token: str

class UserResponse(BaseModel):
    id: str
    username: str
    email: Optional[str]
    role: str
    is_active: bool
    created_at: Optional[str]
    avatar_url: Optional[str] = None
    display_name: Optional[str] = None
    # Nextcloud integration — username override (defaults to OIDC
    # preferred_username) and a flag for whether the user has stored
    # an app password. Cleartext passwords are never serialized.
    nextcloud_username: Optional[str] = None
    has_nextcloud_app_password: bool = False


class UpdateMeRequest(BaseModel):
    """PATCH /me payload. Every field is optional — only what's set
    gets touched. Setting `nextcloud_app_password` to "" clears it."""
    nextcloud_username: Optional[str] = None
    nextcloud_app_password: Optional[str] = None

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

class ChangePasswordRequest(BaseModel):
    current_password: str
    new_password: str


class OidcConfig(BaseModel):
    enabled: bool
    label: str
    login_url: str


class AuthConfigResponse(BaseModel):
    # None when OIDC is disabled / not configured — the frontend uses
    # that to hide the SSO button.
    oidc: Optional[OidcConfig] = None


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def serialize_user(user: User) -> UserResponse:
    """Build the user-facing payload, computing the avatar URL with the
    `provider picture > Gravatar > None` fallback chain."""
    avatar = user.avatar_url or gravatar_url(user.email)
    return UserResponse(
        id=user.id,
        username=user.username,
        email=user.email,
        role=user.role,
        is_active=user.is_active,
        created_at=user.created_at.isoformat() if user.created_at else None,
        avatar_url=avatar,
        display_name=user.display_name,
        nextcloud_username=user.nextcloud_username,
        has_nextcloud_app_password=bool(user.nextcloud_app_password_enc),
    )


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------

@router.get("/config", response_model=AuthConfigResponse)
async def auth_config():
    """Public — tells the frontend which login options to render.

    Returns `oidc: null` when OIDC is disabled or not fully configured,
    so the login page can hide the SSO button without a round-trip to
    the IdP. The `login_url` is browser-navigable (full redirect); it
    starts the Authlib flow that sets the PKCE cookie.
    """
    if not oidc_is_enabled():
        return AuthConfigResponse(oidc=None)
    return AuthConfigResponse(
        oidc=OidcConfig(
            enabled=True,
            label=provider_label(),
            login_url="/api/v1/auth/oidc/login",
        )
    )


@router.post("/login", response_model=TokenResponse)
async def login(body: LoginRequest, db: AsyncSession = Depends(get_db)):
    """Authenticate with username + password, receive JWT tokens."""
    result = await db.execute(
        select(User).where(User.username == body.username)
    )
    user = result.scalar_one_or_none()

    # OIDC-only users (hashed_password IS NULL) can't sign in via this
    # endpoint; they must go through the SSO flow. Treat as auth failure
    # so we don't leak account existence.
    if (
        user is None
        or not user.hashed_password
        or not verify_password(body.password, user.hashed_password)
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
        )
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Account is deactivated",
        )

    return TokenResponse(
        access_token=create_access_token(user.id, user.role),
        refresh_token=create_refresh_token(user.id),
    )


@router.post("/refresh", response_model=TokenResponse)
async def refresh_token(body: RefreshRequest, db: AsyncSession = Depends(get_db)):
    """Exchange a valid refresh token for a new access + refresh pair."""
    try:
        payload = decode_token(body.refresh_token)
        if payload.get("type") != "refresh":
            raise ValueError("not a refresh token")
        user_id = payload["sub"]
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired refresh token",
        )

    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found or deactivated",
        )

    return TokenResponse(
        access_token=create_access_token(user.id, user.role),
        refresh_token=create_refresh_token(user.id),
    )


@router.get("/me", response_model=UserResponse)
async def get_me(current_user: User = Depends(get_current_user)):
    """Return the authenticated user's profile."""
    return serialize_user(current_user)


_NC_USERNAME_RE = re.compile(r"^[a-zA-Z0-9._@-]{1,64}$")


@router.patch("/me", response_model=UserResponse)
async def update_me(
    body: UpdateMeRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Update the authenticated user's Nextcloud integration settings.

    `nextcloud_username` overrides the OIDC `preferred_username` default
    so e.g. the local mule-image user `dtoro` can map to Nextcloud user
    `admin`. `nextcloud_app_password` is encrypted at rest via the
    Fernet helper in `services/secrets.py`; passing an empty string
    clears it.
    """
    from app.services.secrets import encrypt

    changed = False
    if body.nextcloud_username is not None:
        candidate = body.nextcloud_username.strip()
        if candidate and not _NC_USERNAME_RE.match(candidate):
            raise HTTPException(status_code=400, detail="Invalid Nextcloud username")
        current_user.nextcloud_username = candidate or None
        changed = True

    if body.nextcloud_app_password is not None:
        if body.nextcloud_app_password == "":
            current_user.nextcloud_app_password_enc = None
        else:
            current_user.nextcloud_app_password_enc = encrypt(body.nextcloud_app_password)
        changed = True

    if changed:
        await db.commit()
        await db.refresh(current_user)
    return serialize_user(current_user)


@router.post("/change-password")
async def change_password(
    body: ChangePasswordRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Change the authenticated user's password."""
    if not current_user.hashed_password or not verify_password(
        body.current_password, current_user.hashed_password
    ):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Current password is incorrect",
        )
    current_user.hashed_password = hash_password(body.new_password)
    await db.commit()
    return {"status": "ok"}


@router.post("/setup", response_model=TokenResponse, status_code=201)
async def setup(body: SetupRequest, db: AsyncSession = Depends(get_db)):
    """First-run only: create the initial admin account.

    Returns 409 if any user already exists. This endpoint is
    unauthenticated by design — it can only run once.
    """
    count = (await db.execute(select(sa_func.count(User.id)))).scalar()
    if count > 0:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Setup already completed — users exist",
        )

    if len(body.username.strip()) < 2:
        raise HTTPException(status_code=400, detail="Username must be at least 2 characters")
    if len(body.password) < 6:
        raise HTTPException(status_code=400, detail="Password must be at least 6 characters")

    # Every user — including the initial admin — gets their own subfolder
    # under the photo mount root. Nobody owns the root directory itself.
    media_path = os.path.join(settings.photo_dirs, body.username.strip())
    os.makedirs(media_path, exist_ok=True)

    user = User(
        username=body.username.strip(),
        hashed_password=hash_password(body.password),
        role="admin",
        media_path=media_path,
    )
    db.add(user)
    await db.flush()  # get user.id before creating source root

    source_root = SourceRoot(
        name=f"{user.username}'s Library",
        path=media_path,
        user_id=user.id,
    )
    db.add(source_root)
    await db.commit()

    logger.info(f"Initial admin account created: {user.username}")

    return TokenResponse(
        access_token=create_access_token(user.id, user.role),
        refresh_token=create_refresh_token(user.id),
    )


@router.get("/status")
async def auth_status(db: AsyncSession = Depends(get_db)):
    """Public endpoint: returns whether setup has been completed.

    The frontend calls this to decide whether to show the setup page
    or the login page.
    """
    count = (await db.execute(select(sa_func.count(User.id)))).scalar()
    return {"setup_completed": count > 0}


# ---------------------------------------------------------------------------
# OIDC (Authentik) sign-in
# ---------------------------------------------------------------------------

_USERNAME_SANITIZER = re.compile(r"[^a-zA-Z0-9_.-]+")


def _derive_username(claims: dict, existing_usernames: set[str]) -> str:
    """Pick a local username from OIDC claims.

    Order of preference:
      1. `preferred_username` claim (Authentik's usual choice)
      2. local-part of `email`
      3. `sub` claim (always present)

    Strips characters the rest of the app doesn't like in paths/URLs,
    trims to 50 chars (User.username column limit), and appends a short
    suffix on collision so two Authentik users can't land on the same
    local row.
    """
    raw = (
        claims.get("preferred_username")
        or (claims.get("email") or "").split("@", 1)[0]
        or claims.get("sub")
        or "user"
    )
    base = _USERNAME_SANITIZER.sub("", str(raw)).strip("._-") or "user"
    base = base[:40]
    candidate = base
    suffix = 0
    while candidate in existing_usernames:
        suffix += 1
        candidate = f"{base}-{suffix}"[:50]
    return candidate


def _frontend_origin(request: Request) -> str:
    """Best guess at where the SPA lives so the callback redirect lands
    back on the app origin. Uses the configured redirect URI's scheme +
    host (strips /api/... path) when available, falling back to the
    request's own origin."""
    if settings.oidc_redirect_uri:
        parsed = urlparse(settings.oidc_redirect_uri)
        return f"{parsed.scheme}://{parsed.netloc}"
    return f"{request.url.scheme}://{request.url.netloc}"


@router.get("/oidc/login")
async def oidc_login(request: Request):
    """Start the OIDC flow — redirect to Authentik's authorization URL."""
    if not oidc_is_enabled():
        raise HTTPException(status_code=404, detail="OIDC login is not enabled")
    client = get_oidc_client()
    if client is None:
        raise HTTPException(status_code=500, detail="OIDC client not configured")

    redirect_uri = settings.oidc_redirect_uri
    return await client.authorize_redirect(request, redirect_uri)


@router.get("/oidc/callback")
async def oidc_callback(request: Request, db: AsyncSession = Depends(get_db)):
    """Handle the OIDC redirect — exchange code, provision/link user,
    issue our own JWTs, bounce back to the SPA."""
    if not oidc_is_enabled():
        raise HTTPException(status_code=404, detail="OIDC login is not enabled")
    client = get_oidc_client()
    if client is None:
        raise HTTPException(status_code=500, detail="OIDC client not configured")

    try:
        token = await client.authorize_access_token(request)
    except Exception as exc:
        logger.warning("OIDC callback: authorize_access_token failed: %s", exc)
        return _oidc_error_redirect(request, "oidc_exchange_failed")

    # `parse_id_token` verifies signature + nonce; `userinfo` fills in
    # claims some IdPs don't put in the ID token (e.g. picture). We
    # merge both, preferring userinfo when both are present.
    claims = dict(token.get("userinfo") or {})
    if not claims:
        try:
            claims = dict(await client.userinfo(token=token))
        except Exception:
            claims = {}
    id_token_claims = token.get("id_token_claims") or {}
    for k, v in id_token_claims.items():
        claims.setdefault(k, v)

    sub = claims.get("sub")
    if not sub:
        logger.warning("OIDC callback: claims missing `sub` — %r", claims)
        return _oidc_error_redirect(request, "oidc_missing_sub")

    issuer = claims.get("iss") or (settings.oidc_issuer or "").rstrip("/")
    email = claims.get("email")
    display_name = claims.get("name") or claims.get("preferred_username")
    picture = claims.get("picture")
    groups = claims.get("groups") or []
    if isinstance(groups, str):
        groups = [groups]

    admin_groups = set(settings.oidc_admin_group_list)
    role = "admin" if admin_groups and admin_groups.intersection(groups) else "user"

    # 1. Match by (issuer, sub) first — stable identity key.
    user = (await db.execute(
        select(User).where(
            User.oidc_issuer == issuer,
            User.oidc_sub == sub,
        )
    )).scalar_one_or_none()

    # 2. Fall back to email so a pre-existing local account can be
    #    linked on first SSO login (homelab admin keeps their row).
    if user is None and email:
        user = (await db.execute(
            select(User).where(User.email == email)
        )).scalar_one_or_none()

    # 3. Last-resort link by preferred_username. Off by default; only
    #    used in trusted single-tenant setups where local accounts
    #    predate OIDC and never collected email (the app has no UI for
    #    it). Guarded by OIDC_LINK_BY_USERNAME to avoid hijacking
    #    accounts in shared instances.
    if user is None and settings.oidc_link_by_username:
        preferred = claims.get("preferred_username")
        if preferred:
            user = (await db.execute(
                select(User).where(User.username == preferred)
            )).scalar_one_or_none()
            if user is not None:
                logger.info(
                    "OIDC linked existing user %s by preferred_username",
                    preferred,
                )

    if user is None:
        if not settings.oidc_allow_signup:
            logger.info("OIDC signup disabled — rejecting unknown sub=%s email=%s", sub, email)
            return _oidc_error_redirect(request, "oidc_signup_disabled")

        # JIT provision.
        existing = {
            u for (u,) in (await db.execute(select(User.username))).all()
        }
        username = _derive_username(claims, existing)
        media_path = os.path.join(settings.photo_dirs, username)
        os.makedirs(media_path, exist_ok=True)

        user = User(
            username=username,
            email=email,
            hashed_password=None,
            role=role,
            is_active=True,
            media_path=media_path,
            oidc_issuer=issuer,
            oidc_sub=sub,
            avatar_url=picture,
            display_name=display_name,
            # Default the Nextcloud username from preferred_username so
            # the common case "same name on both sides" needs zero
            # configuration. Override is exposed in Settings for the
            # mismatch case (e.g. authentik dtoro ↔ Nextcloud admin).
            nextcloud_username=(claims.get("preferred_username") or None),
        )
        db.add(user)
        await db.flush()

        db.add(SourceRoot(
            name=f"{user.username}'s Library",
            path=media_path,
            user_id=user.id,
        ))
        await db.commit()
        logger.info("OIDC JIT-created user %s (role=%s)", user.username, role)
    else:
        # Refresh profile bits + link identity if needed. We do *not*
        # demote admins created locally; only touch role when admin
        # group mapping is configured.
        changed = False
        if user.oidc_sub != sub or user.oidc_issuer != issuer:
            user.oidc_issuer = issuer
            user.oidc_sub = sub
            changed = True
        if email and user.email != email:
            user.email = email
            changed = True
        if display_name and user.display_name != display_name:
            user.display_name = display_name
            changed = True
        if picture and user.avatar_url != picture:
            user.avatar_url = picture
            changed = True
        # Backfill nextcloud_username on first OIDC login for users that
        # predate the column. NEVER overwrites a value the user already
        # set in Settings — once the override is non-null, it wins.
        if not user.nextcloud_username:
            preferred = claims.get("preferred_username")
            if preferred:
                user.nextcloud_username = preferred
                changed = True
        if admin_groups:
            new_role = "admin" if admin_groups.intersection(groups) else "user"
            if user.role != new_role:
                user.role = new_role
                changed = True
        if not user.is_active:
            # Don't resurrect a deactivated account — surface an error.
            logger.info("OIDC login rejected — user %s is deactivated", user.username)
            return _oidc_error_redirect(request, "oidc_deactivated")
        if changed:
            await db.commit()

    # Mint our own JWTs and bounce back to the SPA. Tokens ride in the
    # URL fragment-free for simplicity; the frontend callback page
    # strips them from the location bar immediately.
    access = create_access_token(user.id, user.role)
    refresh = create_refresh_token(user.id)

    params = urlencode({"access_token": access, "refresh_token": refresh})
    target = f"{_frontend_origin(request)}/auth/callback?{params}"
    return RedirectResponse(url=target, status_code=302)


def _oidc_error_redirect(request: Request, code: str) -> RedirectResponse:
    """Bounce back to the SPA with an `error=` query so the login page
    can render something meaningful instead of a stack trace."""
    target = f"{_frontend_origin(request)}/auth/callback?error={code}"
    return RedirectResponse(url=target, status_code=302)