Gan, Jimmy
152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
import json
|
|
import os
|
|
import threading
|
|
import tempfile
|
|
from typing import Optional
|
|
from pydantic import BaseModel
|
|
from fastapi import HTTPException, Request, status, Depends
|
|
|
|
import config
|
|
|
|
RBAC_FILE = 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:
|
|
if os.path.exists(RBAC_FILE):
|
|
with open(RBAC_FILE) 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(RBAC_FILE, data)
|
|
return result
|
|
|
|
|
|
def save_rbac(data: dict):
|
|
_atomic_json_write(RBAC_FILE, data)
|
|
|
|
|
|
def resolve_role(groups: list[str]) -> Optional[str]:
|
|
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
|
|
|
|
|
|
async def _inject_user(request: Request):
|
|
"""Dependency to store user in request.state for rbac dependencies.
|
|
|
|
Import auth module at runtime to avoid circular imports.
|
|
"""
|
|
import auth as auth_module
|
|
user = await auth_module.get_current_user(request)
|
|
request.state.user = user
|
|
return user
|