Gan, Jimmy
9992105b49
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 1m14s
- Add rbac.py with User model, LDAP group-to-role mapping, page/write dependencies - Proxy auth reads Remote-Groups header to resolve role from LDAP groups - JWT tokens carry role claim, /me returns role + allowed pages - Per-router page access control and viewer write protection - Terminal WebSocket RBAC check - Frontend filters sidebar links and routes by allowed pages - Admin-only Settings page and RBAC override endpoints - Mount rbac.json in docker-compose for page config persistence
95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
import json
|
|
import os
|
|
from typing import Optional
|
|
from pydantic import BaseModel
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
import config
|
|
|
|
RBAC_FILE = os.path.join(config.VOLUME_ROOT, "docker/nas-dashboard/rbac.json")
|
|
|
|
ROLE_GROUP_MAP = {
|
|
"dashboard_admin": "admin",
|
|
"dashboard_member": "member",
|
|
"dashboard_viewer": "viewer",
|
|
}
|
|
|
|
DEFAULT_RBAC = {
|
|
"role_defaults": {
|
|
"admin": {"pages": "*"},
|
|
"member": {"pages": ["dashboard", "docker", "files", "openclaw", "chat-digest"]},
|
|
"viewer": {"pages": ["dashboard"]},
|
|
},
|
|
"user_overrides": {},
|
|
}
|
|
|
|
|
|
class User(BaseModel):
|
|
username: str
|
|
role: str
|
|
pages: list | str # "*" or list of page 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 save_rbac(data: dict):
|
|
os.makedirs(os.path.dirname(RBAC_FILE), exist_ok=True)
|
|
with open(RBAC_FILE, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
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 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
|