nas-tools/dashboard/backend/routers/auth.py

"""Core authentication endpoints: login, token refresh, user info, proxy auth."""

import asyncio
import logging
from datetime import timedelta

import httpx
import jwt
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel
from slowapi import Limiter
from slowapi.util import get_remote_address

import auth_service as auth
import config

logger = logging.getLogger(__name__)
router = APIRouter()
limiter = Limiter(key_func=get_remote_address)


class LoginRequest(BaseModel):
    username: str
    password: str
    totp_code: str = ""


class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str
    user: str
    has_2fa: bool
    role: str = "admin"


class RefreshRequest(BaseModel):
    refresh_token: str = ""


def _set_auth_response_tokens(response: Response, access_token: str, refresh_token: str):
    auth.set_auth_cookies(response, access_token, refresh_token)


def _extract_refresh_token(request: Request, req: RefreshRequest | None = None) -> str:
    cookie_token = request.cookies.get(auth.REFRESH_COOKIE_NAME, "")
    if cookie_token:
        return cookie_token
    if req and req.refresh_token:
        return req.refresh_token
    raise HTTPException(status_code=401, detail="Missing refresh token")


async def _refresh_tokens_from_value(refresh_token_value: str):
    try:
        payload = jwt.decode(refresh_token_value, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")
        username = payload.get("sub")
        role = payload.get("role", "admin")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid user")
        if not auth.verify_token_version(payload.get("tv", -1)):
            raise HTTPException(status_code=401, detail="Token revoked")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    auth.increment_token_version()
    access_token = auth.create_access_token(
        data={"sub": username, "role": role},
        expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = auth.create_refresh_token(data={"sub": username, "role": role})
    return access_token, refresh_token


async def send_failed_login_alert(ip_address: str, username: str, reason: str):
    logger.info("Triggering Telegram alert for %s from %s: %s", username, ip_address, reason)
    if not config.TELEGRAM_BOT_TOKEN or not config.TELEGRAM_CHAT_ID:
        logger.warning("Skipping Telegram alert: missing config")
        return
    msg = f"🚨 *Dashboard Security Alert* 🚨\nFailed login attempt detected.\n\n👤 User: `{username}`\n🌐 IP: `{ip_address}`\n❌ Reason: {reason}"
    try:
        client_options = {}
        if config.TELEGRAM_PROXY:
            client_options["proxy"] = config.TELEGRAM_PROXY

        async with httpx.AsyncClient(**client_options) as client:
            res = await client.post(
                f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage",
                data={"chat_id": config.TELEGRAM_CHAT_ID, "text": msg, "parse_mode": "Markdown"},
                timeout=10.0,
            )
            logger.info("Telegram alert sent, status: %s", res.status_code)
    except Exception as e:
        logger.warning("Failed to send Telegram alert: %s", e)


@router.post("/login", response_model=Token)
@limiter.limit("5/minute")
async def login(creds: LoginRequest, request: Request, response: Response):
    client_ip = request.client.host
    # Verify username/password
    if creds.username != config.ADMIN_USER or not auth.verify_password(creds.password, auth.load_password_hash()):
        asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Incorrect username or password"))
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    # Check 2FA
    totp_secret = auth.load_totp_secret()
    if totp_secret:
        if not creds.totp_code:
            asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Missing 2FA Code"))
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="2FA code required",
                headers={"WWW-Authenticate": "Bearer"},
            )
        if not auth.verify_totp(creds.totp_code, totp_secret):
            asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Invalid 2FA Code"))
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )

    # Direct login is admin-only fallback
    access_token = auth.create_access_token(
        data={"sub": creds.username, "role": "admin"},
        expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = auth.create_refresh_token(data={"sub": creds.username, "role": "admin"})
    _set_auth_response_tokens(response, access_token, refresh_token)
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "user": creds.username,
        "has_2fa": bool(totp_secret),
        "role": "admin",
    }


@router.get("/proxy")
async def proxy_auth(request: Request, response: Response):
    """Auto-login when Authelia has already authenticated the user via forward-auth.
    Caddy sets Remote-User and Remote-Groups headers after successful Authelia verification."""
    from rbac import resolve_role

    if not config.is_trusted_proxy(request.client.host):
        logger.warning(f"Rejected proxy auth attempt from unauthorized IP: {request.client.host}")
        raise HTTPException(status_code=403, detail="Unauthorized proxy source")

    remote_user = request.headers.get("Remote-User")
    if not remote_user:
        raise HTTPException(status_code=401, detail="No proxy auth header")

    # Parse groups from Remote-Groups header (comma-separated)
    remote_groups_raw = request.headers.get("Remote-Groups", "")
    groups = [g.strip() for g in remote_groups_raw.split(",") if g.strip()]

    logger.info(f"Proxy auth: user={remote_user}, groups={groups}")
    role = resolve_role(groups)
    logger.info(f"Resolved role: {role}")
    if role is None:
        logger.warning(f"User {remote_user} has no dashboard role (groups: {groups})")
        raise HTTPException(status_code=403, detail="User not authorized for dashboard")

    access_token = auth.create_access_token(
        data={"sub": remote_user, "role": role},
        expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = auth.create_refresh_token(data={"sub": remote_user, "role": role})
    _set_auth_response_tokens(response, access_token, refresh_token)
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "user": remote_user,
        "has_2fa": True,
        "role": role,
    }


@router.get("/me")
async def read_users_me(current_user=Depends(auth.get_current_user)):
    totp_secret = auth.load_totp_secret()
    return {
        "username": current_user.username,
        "role": current_user.role,
        "pages": current_user.pages,
        "sidebar_links": current_user.sidebar_links,
        "has_2fa": bool(totp_secret),
    }


@router.post("/refresh")
@limiter.limit("10/minute")
async def refresh_token(request: Request, response: Response, req: RefreshRequest | None = Body(default=None)):
    refresh_token_value = _extract_refresh_token(request, req)
    access_token, refresh_token = await _refresh_tokens_from_value(refresh_token_value)
    _set_auth_response_tokens(response, access_token, refresh_token)
    return {"access_token": access_token, "refresh_token": refresh_token}


@router.post("/logout")
async def logout(response: Response):
    auth.increment_token_version()
    auth.clear_auth_cookies(response)
    return {"ok": True}


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


@router.post("/change-password")
@limiter.limit("5/minute")
async def change_password(req: ChangePasswordRequest, request: Request, current_user=Depends(auth.get_current_user)):
    if not auth.verify_password(req.current_password, auth.load_password_hash()):
        raise HTTPException(status_code=400, detail="Current password is incorrect")
    if len(req.new_password) < 8:
        raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
    auth.save_password_hash(auth.get_password_hash(req.new_password))
    return {"message": "Password changed successfully"}


# RBAC admin endpoints
@router.get("/rbac/config")
async def rbac_config(current_user=Depends(auth.get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    from rbac import load_rbac

    return load_rbac()


@router.put("/rbac/overrides/{username}")
async def rbac_set_override(username: str, request: Request, current_user=Depends(auth.get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    from rbac import update_rbac

    body = await request.json()
    pages = body.get("pages")
    if pages is None:
        raise HTTPException(status_code=400, detail="Missing pages field")

    def mutator(rbac):
        existing = rbac.setdefault("user_overrides", {}).get(username, {})
        existing["pages"] = pages
        rbac["user_overrides"][username] = existing

    update_rbac(mutator)
    return {"ok": True}


@router.get("/preferences")
async def get_preferences(current_user=Depends(auth.get_current_user)):
    from rbac import get_sidebar_order

    order = get_sidebar_order(current_user.username)
    return {"sidebar_order": order}


@router.put("/preferences")
async def save_preferences(request: Request, current_user=Depends(auth.get_current_user)):
    import traceback

    from rbac import save_sidebar_order

    try:
        body = await request.json()
        order = body.get("sidebar_order")
        if order is None or not isinstance(order, dict):
            raise HTTPException(status_code=400, detail="Missing sidebar_order dict")
        save_sidebar_order(current_user.username, order)
        return {"ok": True}
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("Error saving preferences for user %s", current_user.username)
        raise HTTPException(status_code=500, detail="Failed to save preferences")


@router.delete("/rbac/overrides/{username}")
async def rbac_delete_override(username: str, current_user=Depends(auth.get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    from rbac import update_rbac

    def mutator(rbac):
        rbac.get("user_overrides", {}).pop(username, None)

    update_rbac(mutator)
    return {"ok": True}


@router.put("/rbac/roles/{role}")
async def rbac_update_role(role: str, request: Request, current_user=Depends(auth.get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    from rbac import update_rbac

    body = await request.json()
    pages = body.get("pages")
    sidebar_links = body.get("sidebar_links")
    if pages is None:
        raise HTTPException(status_code=400, detail="Missing pages field")
    if pages != "*" and not isinstance(pages, list):
        raise HTTPException(status_code=400, detail="pages must be '*' or a list")
    if sidebar_links is not None and sidebar_links != "*" and not isinstance(sidebar_links, list):
        raise HTTPException(status_code=400, detail="sidebar_links must be '*' or a list")

    def mutator(rbac):
        role_config = {"pages": pages}
        if sidebar_links is not None:
            role_config["sidebar_links"] = sidebar_links
        rbac.setdefault("role_defaults", {})[role] = role_config

    update_rbac(mutator)
    return {"ok": True}