nas-tools/dashboard/backend/rbac.py

import json
import os
import tempfile
import threading

from fastapi import HTTPException, Request
from pydantic import BaseModel

import config


def get_rbac_file():
    return os.path.join(config.VOLUME_ROOT, "docker/nas-dashboard/rbac.json")


_rbac_lock = threading.RLock()

ROLE_GROUP_MAP = {
    "dashboard_admin": "admin",
    "dashboard_member": "member",
    "dashboard_viewer": "viewer",
    "admins": "admin",  # Authelia admin group
    "users": "member",  # Authelia user group
}

DEFAULT_RBAC = {
    "role_defaults": {
        "admin": {"pages": "*", "sidebar_links": "*"},
        "member": {"pages": ["dashboard", "docker", "files", "openclaw", "chat-digest", "opc"], "sidebar_links": "*"},
        "viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]},
    },
    "user_overrides": {},
}


class User(BaseModel):
    username: str
    role: str
    pages: list | str  # "*" or list of page ids
    sidebar_links: list | str = "*"  # "*" or list of sidebar link labels/ids


def load_rbac() -> dict:
    try:
        rbac_path = get_rbac_file()
        if os.path.exists(rbac_path):
            with open(rbac_path) as f:
                return json.load(f)
    except Exception:
        pass
    return DEFAULT_RBAC.copy()


def _atomic_json_write(path: str, data: dict):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(path), prefix=".tmp-rbac-", suffix=".json")
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(data, f, indent=2)
            f.flush()
            os.fsync(f.fileno())
        os.replace(temp_path, path)
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)


def update_rbac(mutator):
    with _rbac_lock:
        data = load_rbac()
        result = mutator(data)
        _atomic_json_write(get_rbac_file(), data)
        return result


def save_rbac(data: dict):
    _atomic_json_write(get_rbac_file(), data)


def resolve_role(groups: list[str]) -> str | None:
    for group in groups:
        if group in ROLE_GROUP_MAP:
            return ROLE_GROUP_MAP[group]
    return None


def get_pages(username: str, role: str) -> list | str:
    rbac = load_rbac()
    overrides = rbac.get("user_overrides", {})
    if username in overrides:
        return overrides[username].get("pages", rbac["role_defaults"].get(role, {}).get("pages", []))
    return rbac.get("role_defaults", {}).get(role, {}).get("pages", [])


def get_sidebar_links(username: str, role: str) -> list | str:
    rbac = load_rbac()
    overrides = rbac.get("user_overrides", {})
    if username in overrides and "sidebar_links" in overrides[username]:
        return overrides[username]["sidebar_links"]
    return rbac.get("role_defaults", {}).get(role, {}).get("sidebar_links", "*")


def get_sidebar_order(username: str) -> dict | None:
    rbac = load_rbac()
    overrides = rbac.get("user_overrides", {})
    return overrides.get(username, {}).get("sidebar_order")


def save_sidebar_order(username: str, order: dict):
    def mutator(rbac):
        rbac.setdefault("user_overrides", {}).setdefault(username, {})["sidebar_order"] = order

    update_rbac(mutator)


def has_page_access(user: User, page: str) -> bool:
    if user.pages == "*":
        return True
    return page in user.pages


def require_page(page: str):
    """FastAPI dependency factory — 403 if user lacks page access."""

    async def checker(request: Request):
        user: User = request.state.user
        if not has_page_access(user, page):
            raise HTTPException(status_code=403, detail="Access denied")

    return checker


def require_admin():
    """FastAPI dependency — 403 if user is not admin."""

    async def checker(request: Request):
        user: User = request.state.user
        if user.role != "admin":
            raise HTTPException(status_code=403, detail="Admin only")

    return checker


def require_write():
    """FastAPI dependency — 403 if viewer tries non-GET method."""

    async def checker(request: Request):
        user: User = request.state.user
        if user.role == "viewer" and request.method != "GET":
            raise HTTPException(status_code=403, detail="Read-only access")

    return checker