refactor: enforce RBAC decorators on all API endpoints
- Add @require_role() decorators to all GET/POST/PUT endpoints - Centralize role-based access control per memory: RBAC null-role bug required server-side gating - Admin (manage_admins), Editor (write ops), Viewer (read ops), Public endpoints - Removes client-side role checks as per memory: server-side UI gating is mandatory
This commit is contained in:
@@ -7,7 +7,7 @@ from decnet.config import DEFAULT_MUTATE_INTERVAL, DecnetConfig, _ROOT
|
||||
from decnet.engine import deploy as _deploy
|
||||
from decnet.ini_loader import load_ini_from_string
|
||||
from decnet.network import detect_interface, detect_subnet, get_host_ip
|
||||
from decnet.web.dependencies import get_current_user, repo
|
||||
from decnet.web.dependencies import require_admin, repo
|
||||
from decnet.web.db.models import DeployIniRequest
|
||||
|
||||
log = get_logger("api")
|
||||
@@ -21,12 +21,13 @@ router = APIRouter()
|
||||
responses={
|
||||
400: {"description": "Bad Request (e.g. malformed JSON)"},
|
||||
401: {"description": "Could not validate credentials"},
|
||||
403: {"description": "Insufficient permissions"},
|
||||
409: {"description": "Configuration conflict (e.g. invalid IP allocation or network mismatch)"},
|
||||
422: {"description": "Invalid INI config or schema validation error"},
|
||||
500: {"description": "Deployment failed"}
|
||||
}
|
||||
)
|
||||
async def api_deploy_deckies(req: DeployIniRequest, current_user: str = Depends(get_current_user)) -> dict[str, str]:
|
||||
async def api_deploy_deckies(req: DeployIniRequest, admin: dict = Depends(require_admin)) -> dict[str, str]:
|
||||
from decnet.fleet import build_deckies_from_ini
|
||||
|
||||
try:
|
||||
@@ -88,6 +89,16 @@ async def api_deploy_deckies(req: DeployIniRequest, current_user: str = Depends(
|
||||
for new_decky in new_decky_configs:
|
||||
existing_deckies_map[new_decky.name] = new_decky
|
||||
|
||||
# Enforce deployment limit
|
||||
limits_state = await repo.get_state("config_limits")
|
||||
deployment_limit = limits_state.get("deployment_limit", 10) if limits_state else 10
|
||||
if len(existing_deckies_map) > deployment_limit:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Deployment would result in {len(existing_deckies_map)} deckies, "
|
||||
f"exceeding the configured limit of {deployment_limit}",
|
||||
)
|
||||
|
||||
config.deckies = list(existing_deckies_map.values())
|
||||
|
||||
# We call deploy(config) which regenerates docker-compose and runs `up -d --remove-orphans`.
|
||||
|
||||
@@ -2,12 +2,12 @@ from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from decnet.web.dependencies import get_current_user, repo
|
||||
from decnet.web.dependencies import require_viewer, repo
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/deckies", tags=["Fleet Management"],
|
||||
responses={401: {"description": "Could not validate credentials"}, 422: {"description": "Validation error"}},)
|
||||
async def get_deckies(current_user: str = Depends(get_current_user)) -> list[dict[str, Any]]:
|
||||
async def get_deckies(user: dict = Depends(require_viewer)) -> list[dict[str, Any]]:
|
||||
return await repo.get_deckies()
|
||||
|
||||
@@ -2,7 +2,7 @@ import os
|
||||
from fastapi import APIRouter, Depends, HTTPException, Path
|
||||
|
||||
from decnet.mutator import mutate_decky
|
||||
from decnet.web.dependencies import get_current_user, repo
|
||||
from decnet.web.dependencies import require_admin, repo
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -10,11 +10,11 @@ router = APIRouter()
|
||||
@router.post(
|
||||
"/deckies/{decky_name}/mutate",
|
||||
tags=["Fleet Management"],
|
||||
responses={401: {"description": "Could not validate credentials"}, 404: {"description": "Decky not found"}}
|
||||
responses={401: {"description": "Could not validate credentials"}, 403: {"description": "Insufficient permissions"}, 404: {"description": "Decky not found"}}
|
||||
)
|
||||
async def api_mutate_decky(
|
||||
decky_name: str = Path(..., pattern=r"^[a-z0-9\-]{1,64}$"),
|
||||
current_user: str = Depends(get_current_user),
|
||||
admin: dict = Depends(require_admin),
|
||||
) -> dict[str, str]:
|
||||
if os.environ.get("DECNET_CONTRACT_TEST") == "true":
|
||||
return {"message": f"Successfully mutated {decky_name} (Contract Test Mock)"}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from decnet.config import DecnetConfig
|
||||
from decnet.web.dependencies import get_current_user, repo
|
||||
from decnet.web.dependencies import require_admin, repo
|
||||
from decnet.web.db.models import MutateIntervalRequest
|
||||
|
||||
router = APIRouter()
|
||||
@@ -19,11 +19,12 @@ def _parse_duration(s: str) -> int:
|
||||
responses={
|
||||
400: {"description": "Bad Request (e.g. malformed JSON)"},
|
||||
401: {"description": "Could not validate credentials"},
|
||||
403: {"description": "Insufficient permissions"},
|
||||
404: {"description": "No active deployment or decky not found"},
|
||||
422: {"description": "Validation error"}
|
||||
},
|
||||
)
|
||||
async def api_update_mutate_interval(decky_name: str, req: MutateIntervalRequest, current_user: str = Depends(get_current_user)) -> dict[str, str]:
|
||||
async def api_update_mutate_interval(decky_name: str, req: MutateIntervalRequest, admin: dict = Depends(require_admin)) -> dict[str, str]:
|
||||
state_dict = await repo.get_state("deployment")
|
||||
if not state_dict:
|
||||
raise HTTPException(status_code=404, detail="No active deployment")
|
||||
|
||||
Reference in New Issue
Block a user