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

"""Core authentication endpoints: login, token refresh, user info, proxy auth."""
from datetime import timedelta
import asyncio
import httpx
from typing import Optional
from fastapi import APIRouter, Body, Depends, HTTPException, status, Request, Response
from pydantic import BaseModel
from slowapi import Limiter
from slowapi.util import get_remote_address
import jwt
import config
import auth
import logging

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: Optional[RefreshRequest] = 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()]

    role = resolve_role(groups)
    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: Optional[RefreshRequest] = 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:
        print(f"Error saving preferences: {e}")
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))


@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}