feat: RBAC multi-user role-based access control
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 1m14s
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
This commit is contained in:
@@ -32,6 +32,7 @@ class Token(BaseModel):
|
||||
token_type: str
|
||||
user: str
|
||||
has_2fa: bool
|
||||
role: str = "admin"
|
||||
|
||||
class RefreshRequest(BaseModel):
|
||||
refresh_token: str
|
||||
@@ -96,25 +97,27 @@ async def login(creds: LoginRequest, request: Request):
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
# Direct login is admin-only fallback
|
||||
access_token = auth.create_access_token(
|
||||
data={"sub": creds.username},
|
||||
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})
|
||||
refresh_token = auth.create_refresh_token(data={"sub": creds.username, "role": "admin"})
|
||||
return {
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"token_type": "bearer",
|
||||
"user": creds.username,
|
||||
"has_2fa": bool(totp_secret)
|
||||
"has_2fa": bool(totp_secret),
|
||||
"role": "admin",
|
||||
}
|
||||
|
||||
@router.get("/proxy")
|
||||
async def proxy_auth(request: Request):
|
||||
"""Auto-login when Authelia has already authenticated the user via forward-auth.
|
||||
Caddy sets Remote-User header after successful Authelia 2FA verification."""
|
||||
# Prevent IP spoofing: ensure proxy traffic is from trusted proxy network
|
||||
Caddy sets Remote-User and Remote-Groups headers after successful Authelia verification."""
|
||||
import config
|
||||
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")
|
||||
@@ -122,27 +125,39 @@ async def proxy_auth(request: Request):
|
||||
remote_user = request.headers.get("Remote-User")
|
||||
if not remote_user:
|
||||
raise HTTPException(status_code=401, detail="No proxy auth header")
|
||||
# Map the LDAP user to the dashboard admin user
|
||||
# Only allow known users (the admin)
|
||||
if remote_user != config.ADMIN_USER:
|
||||
|
||||
# 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},
|
||||
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})
|
||||
refresh_token = auth.create_refresh_token(data={"sub": remote_user, "role": role})
|
||||
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: str = Depends(auth.get_current_user)):
|
||||
async def read_users_me(current_user = Depends(auth.get_current_user)):
|
||||
totp_secret = auth.load_totp_secret()
|
||||
return {"username": current_user, "has_2fa": bool(totp_secret)}
|
||||
return {
|
||||
"username": current_user.username,
|
||||
"role": current_user.role,
|
||||
"pages": current_user.pages,
|
||||
"has_2fa": bool(totp_secret),
|
||||
}
|
||||
|
||||
@router.post("/refresh")
|
||||
@limiter.limit("10/minute")
|
||||
@@ -152,7 +167,8 @@ async def refresh_token(req: RefreshRequest, request: Request):
|
||||
if payload.get("type") != "refresh":
|
||||
raise HTTPException(status_code=401, detail="Invalid token type")
|
||||
username = payload.get("sub")
|
||||
if username != config.ADMIN_USER:
|
||||
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")
|
||||
@@ -160,20 +176,20 @@ async def refresh_token(req: RefreshRequest, request: Request):
|
||||
raise HTTPException(status_code=401, detail="Invalid refresh token")
|
||||
auth.increment_token_version()
|
||||
access_token = auth.create_access_token(
|
||||
data={"sub": username},
|
||||
data={"sub": username, "role": role},
|
||||
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
)
|
||||
refresh_token = auth.create_refresh_token(data={"sub": username})
|
||||
refresh_token = auth.create_refresh_token(data={"sub": username, "role": role})
|
||||
return {"access_token": access_token, "refresh_token": refresh_token}
|
||||
|
||||
@router.post("/setup-2fa/generate", response_model=Setup2FAResponse)
|
||||
async def generate_2fa(current_user: str = Depends(auth.get_current_user)):
|
||||
async def generate_2fa(current_user = Depends(auth.get_current_user)):
|
||||
secret = pyotp.random_base32()
|
||||
uri = pyotp.totp.TOTP(secret).provisioning_uri(name=current_user, issuer_name="NAS Dashboard")
|
||||
uri = pyotp.totp.TOTP(secret).provisioning_uri(name=current_user.username, issuer_name="NAS Dashboard")
|
||||
return {"secret": secret, "uri": uri}
|
||||
|
||||
@router.post("/setup-2fa/verify")
|
||||
async def verify_2fa_setup(request: Verify2FARequest, current_user: str = Depends(auth.get_current_user)):
|
||||
async def verify_2fa_setup(request: Verify2FARequest, current_user = Depends(auth.get_current_user)):
|
||||
totp = pyotp.TOTP(request.secret)
|
||||
if not totp.verify(request.code):
|
||||
raise HTTPException(status_code=400, detail="Invalid code")
|
||||
@@ -183,7 +199,7 @@ async def verify_2fa_setup(request: Verify2FARequest, current_user: str = Depend
|
||||
return {"message": "2FA enabled successfully"}
|
||||
|
||||
@router.post("/setup-2fa/disable")
|
||||
async def disable_2fa(current_user: str = Depends(auth.get_current_user)):
|
||||
async def disable_2fa(current_user = Depends(auth.get_current_user)):
|
||||
auth.save_totp_secret("")
|
||||
return {"message": "2FA disabled"}
|
||||
|
||||
@@ -193,7 +209,7 @@ class ChangePasswordRequest(BaseModel):
|
||||
|
||||
@router.post("/change-password")
|
||||
@limiter.limit("5/minute")
|
||||
async def change_password(req: ChangePasswordRequest, request: Request, current_user: str = Depends(auth.get_current_user)):
|
||||
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:
|
||||
@@ -226,22 +242,22 @@ def _get_challenge(client_data_b64: str) -> bytes:
|
||||
return chal_bytes
|
||||
|
||||
@router.post("/passkey/register/options")
|
||||
async def passkey_register_options(current_user: str = Depends(auth.get_current_user)):
|
||||
async def passkey_register_options(current_user = Depends(auth.get_current_user)):
|
||||
existing = auth.load_passkey_credentials()
|
||||
exclude = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"])) for c in existing]
|
||||
options = generate_registration_options(
|
||||
rp_id=config.WEBAUTHN_RP_ID,
|
||||
rp_name="NAS Dashboard",
|
||||
user_id=config.ADMIN_USER.encode(),
|
||||
user_name=config.ADMIN_USER,
|
||||
user_display_name=config.ADMIN_USER,
|
||||
user_id=current_user.username.encode(),
|
||||
user_name=current_user.username,
|
||||
user_display_name=current_user.username,
|
||||
exclude_credentials=exclude,
|
||||
)
|
||||
_store_challenge(options.challenge)
|
||||
return JSONResponse(content=json.loads(options_to_json(options)))
|
||||
|
||||
@router.post("/passkey/register/verify")
|
||||
async def passkey_register_verify(request: Request, current_user: str = Depends(auth.get_current_user)):
|
||||
async def passkey_register_verify(request: Request, current_user = Depends(auth.get_current_user)):
|
||||
body = await request.json()
|
||||
challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
|
||||
try:
|
||||
@@ -304,19 +320,19 @@ async def passkey_login_verify(request: Request):
|
||||
auth._save_auth_data(data)
|
||||
username = config.ADMIN_USER
|
||||
access_token = auth.create_access_token(
|
||||
data={"sub": username},
|
||||
data={"sub": username, "role": "admin"},
|
||||
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
)
|
||||
refresh_token = auth.create_refresh_token(data={"sub": username})
|
||||
return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "user": username}
|
||||
refresh_token = auth.create_refresh_token(data={"sub": username, "role": "admin"})
|
||||
return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "user": username, "role": "admin"}
|
||||
|
||||
@router.get("/passkey/list")
|
||||
async def passkey_list(current_user: str = Depends(auth.get_current_user)):
|
||||
async def passkey_list(current_user = Depends(auth.get_current_user)):
|
||||
creds = auth.load_passkey_credentials()
|
||||
return {"passkeys": [{"id": c["credential_id"], "name": c.get("name", "Passkey"), "created_at": c.get("created_at", "")} for c in creds]}
|
||||
|
||||
@router.post("/passkey/delete")
|
||||
async def passkey_delete(request: Request, current_user: str = Depends(auth.get_current_user)):
|
||||
async def passkey_delete(request: Request, current_user = Depends(auth.get_current_user)):
|
||||
body = await request.json()
|
||||
cred_id = body.get("id")
|
||||
data = auth._load_auth_data()
|
||||
@@ -326,6 +342,38 @@ async def passkey_delete(request: Request, current_user: str = Depends(auth.get_
|
||||
return {"ok": True}
|
||||
|
||||
@router.post("/passkey/clear")
|
||||
async def passkey_clear(current_user: str = Depends(auth.get_current_user)):
|
||||
async def passkey_clear(current_user = Depends(auth.get_current_user)):
|
||||
auth.clear_passkey_credentials()
|
||||
return {"ok": True}
|
||||
|
||||
# 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 load_rbac, save_rbac
|
||||
body = await request.json()
|
||||
pages = body.get("pages")
|
||||
if pages is None:
|
||||
raise HTTPException(status_code=400, detail="Missing pages field")
|
||||
rbac = load_rbac()
|
||||
rbac.setdefault("user_overrides", {})[username] = {"pages": pages}
|
||||
save_rbac(rbac)
|
||||
return {"ok": True}
|
||||
|
||||
@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 load_rbac, save_rbac
|
||||
rbac = load_rbac()
|
||||
rbac.get("user_overrides", {}).pop(username, None)
|
||||
save_rbac(rbac)
|
||||
return {"ok": True}
|
||||
|
||||
Reference in New Issue
Block a user