Merge dev to main: Security improvements and comprehensive test infrastructure #39
@@ -8,7 +8,9 @@ from fastapi import HTTPException, Request, status, Depends
|
||||
|
||||
import config
|
||||
|
||||
RBAC_FILE = os.path.join(config.VOLUME_ROOT, "docker/nas-dashboard/rbac.json")
|
||||
def get_rbac_file():
|
||||
return os.path.join(config.VOLUME_ROOT, "docker/nas-dashboard/rbac.json")
|
||||
|
||||
_rbac_lock = threading.RLock()
|
||||
|
||||
ROLE_GROUP_MAP = {
|
||||
@@ -38,8 +40,9 @@ class User(BaseModel):
|
||||
|
||||
def load_rbac() -> dict:
|
||||
try:
|
||||
if os.path.exists(RBAC_FILE):
|
||||
with open(RBAC_FILE) as f:
|
||||
rbac_path = get_rbac_file()
|
||||
if os.path.exists(rbac_path):
|
||||
with open(rbac_path) as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -64,12 +67,12 @@ def update_rbac(mutator):
|
||||
with _rbac_lock:
|
||||
data = load_rbac()
|
||||
result = mutator(data)
|
||||
_atomic_json_write(RBAC_FILE, data)
|
||||
_atomic_json_write(get_rbac_file(), data)
|
||||
return result
|
||||
|
||||
|
||||
def save_rbac(data: dict):
|
||||
_atomic_json_write(RBAC_FILE, data)
|
||||
_atomic_json_write(get_rbac_file(), data)
|
||||
|
||||
|
||||
def resolve_role(groups: list[str]) -> Optional[str]:
|
||||
|
||||
Reference in New Issue
Block a user